Merge remote-tracking branch 'origin/main'

fluxer_api/src/api/channel/services/CallService.test.ts

@@ -0,0 +1,210 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {createChannelID, createUserID, type UserID} from '@app/api/BrandedTypes';
+import type {IChannelRepository} from '@app/api/channel/IChannelRepository';
+import {CallService} from '@app/api/channel/services/CallService';
+import type {IGuildRepositoryAggregate} from '@app/api/guild/repositories/IGuildRepositoryAggregate';
+import type {CallCaller, CallData, IGatewayService} from '@app/api/infrastructure/IGatewayService';
+import type {ISnowflakeService} from '@app/api/infrastructure/ISnowflakeService';
+import type {IVoiceRoomStore} from '@app/api/infrastructure/IVoiceRoomStore';
+import type {UserCacheService} from '@app/api/infrastructure/UserCacheService';
+import type {RequestCache} from '@app/api/middleware/RequestCacheMiddleware';
+import type {Channel} from '@app/api/models/Channel';
+import type {User} from '@app/api/models/User';
+import type {ReadStateService} from '@app/api/read_state/ReadStateService';
+import type {IUserRepository} from '@app/api/user/IUserRepository';
+import {ChannelTypes} from '@fluxer/constants/src/ChannelConstants';
+import type {UserPartialResponse} from '@fluxer/schema/src/domains/user/UserResponseSchemas';
+import {beforeEach, describe, expect, it} from 'vitest';
+
+const CALLER_ID = createUserID(1n);
+const RECIPIENT_ID = createUserID(2n);
+const CHANNEL_ID = createChannelID(12n);
+
+interface CallerOverrides {
+	username?: string;
+	globalName?: string | null;
+	avatar?: string | null;
+	nickname?: string;
+	userRowMissing?: boolean;
+}
+
+interface Harness {
+	service: CallService;
+	created: Array<CallCaller | undefined>;
+	rung: Array<CallCaller | undefined>;
+}
+
+const EXISTING_CALL: CallData = {
+	channel_id: CHANNEL_ID.toString(),
+	message_id: '99',
+	region: 'automatic',
+	ringing: [],
+	recipients: [CALLER_ID.toString(), RECIPIENT_ID.toString()],
+	voice_states: [],
+};
+
+function harness(overrides: CallerOverrides, existingCall: CallData | null): Harness {
+	const username = overrides.username ?? 'elias';
+	const globalName = overrides.globalName === undefined ? 'Elias' : overrides.globalName;
+	const avatar = overrides.avatar === undefined ? 'a1b2c3d4' : overrides.avatar;
+	const nicknames = new Map<string, string>();
+	if (overrides.nickname !== undefined) {
+		nicknames.set(CALLER_ID.toString(), overrides.nickname);
+	}
+	const channel = {
+		id: CHANNEL_ID,
+		type: ChannelTypes.GROUP_DM,
+		recipientIds: new Set<UserID>([CALLER_ID, RECIPIENT_ID]),
+		nicknames,
+	} as unknown as Channel;
+	const created: Array<CallCaller | undefined> = [];
+	const rung: Array<CallCaller | undefined> = [];
+	const channelRepository = {
+		findUnique: async () => channel,
+		upsertMessage: async () => {},
+		getMessage: async () => null,
+	} as unknown as IChannelRepository;
+	const userRepository = {
+		findUnique: async () => (overrides.userRowMissing ? null : ({...callerUser(username, globalName, avatar)} as User)),
+		listUsers: async () => [],
+		findSettings: async () => null,
+		isDmChannelOpen: async () => true,
+	} as unknown as IUserRepository;
+	const gatewayService = {
+		getCall: async () => existingCall,
+		createCall: async (
+			_channelId: unknown,
+			_messageId: string,
+			_region: string,
+			_ringing: Array<string>,
+			_recipients: Array<string>,
+			caller?: CallCaller,
+		) => {
+			created.push(caller);
+			return EXISTING_CALL;
+		},
+		ringCallRecipients: async (_channelId: unknown, _recipients: Array<string>, caller?: CallCaller) => {
+			rung.push(caller);
+			return true;
+		},
+	} as unknown as IGatewayService;
+	const userCacheService = {
+		getUserPartialResponse: async (): Promise<UserPartialResponse> =>
+			({
+				id: CALLER_ID.toString(),
+				username,
+				discriminator: '0001',
+				global_name: globalName,
+				avatar,
+				avatar_color: null,
+				flags: 0,
+			}) as unknown as UserPartialResponse,
+	} as unknown as UserCacheService;
+	const snowflakeService = {
+		generateForChannel: async () => 7777n,
+	} as unknown as ISnowflakeService;
+	const readStateService = {
+		ackMessage: async () => {},
+		bulkIncrementMentionCounts: async () => {},
+	} as unknown as ReadStateService;
+	const service = new CallService(
+		channelRepository,
+		userRepository,
+		{} as unknown as IGuildRepositoryAggregate,
+		gatewayService,
+		userCacheService,
+		snowflakeService,
+		readStateService,
+		null,
+		{} as unknown as IVoiceRoomStore,
+	);
+	return {service, created, rung};
+}
+
+function callerUser(username: string, globalName: string | null, avatar: string | null): Partial<User> {
+	return {
+		id: CALLER_ID,
+		username,
+		globalName,
+		avatarHash: avatar,
+		isBot: false,
+	};
+}
+
+const requestCache = {
+	userPartials: new Map(),
+} as unknown as RequestCache;
+
+describe('CallService caller identity', () => {
+	let harnessState: Harness;
+
+	const createCall = (overrides: CallerOverrides = {}) => {
+		harnessState = harness(overrides, null);
+		return harnessState.service.createOrGetCall({
+			userId: CALLER_ID,
+			channelId: CHANNEL_ID,
+			ringing: [RECIPIENT_ID],
+			requestCache,
+		});
+	};
+
+	const ringExistingCall = (overrides: CallerOverrides = {}) => {
+		harnessState = harness(overrides, EXISTING_CALL);
+		return harnessState.service.ringCallRecipients({
+			userId: CALLER_ID,
+			channelId: CHANNEL_ID,
+			requestCache,
+		});
+	};
+
+	beforeEach(() => {
+		requestCache.userPartials.clear();
+	});
+
+	it('sends the caller id, display name and avatar hash to createCall', async () => {
+		await createCall();
+		expect(harnessState.created).toEqual([{id: '1', name: 'Elias', avatar: 'a1b2c3d4'}]);
+	});
+
+	it('prefers the group dm nickname over the global name on createCall', async () => {
+		await createCall({nickname: 'Eli'});
+		expect(harnessState.created[0]?.name).toBe('Eli');
+	});
+
+	it('falls back to the username when the caller has no nickname and no global name', async () => {
+		await createCall({globalName: null});
+		expect(harnessState.created[0]?.name).toBe('elias');
+	});
+
+	it('sends a null avatar when the caller has no custom avatar', async () => {
+		await createCall({avatar: null});
+		expect(harnessState.created[0]).toEqual({id: '1', name: 'Elias', avatar: null});
+	});
+
+	it('sends no caller at all when the caller user row is gone', async () => {
+		await createCall({userRowMissing: true});
+		expect(harnessState.created).toEqual([undefined]);
+	});
+
+	it('sends the caller id, display name and avatar hash to ringCallRecipients', async () => {
+		await ringExistingCall();
+		expect(harnessState.rung).toEqual([{id: '1', name: 'Elias', avatar: 'a1b2c3d4'}]);
+	});
+
+	it('prefers the group dm nickname over the global name on ringCallRecipients', async () => {
+		await ringExistingCall({nickname: 'Eli'});
+		expect(harnessState.rung[0]?.name).toBe('Eli');
+	});
+
+	it('falls back to the username on ringCallRecipients', async () => {
+		await ringExistingCall({globalName: null});
+		expect(harnessState.rung[0]?.name).toBe('elias');
+	});
+
+	it('resolves the caller on the ring branch and not on the create branch', async () => {
+		await ringExistingCall();
+		expect(harnessState.created).toEqual([]);
+		expect(harnessState.rung).toHaveLength(1);
+	});
+});

fluxer_api/src/api/channel/services/CallService.ts

@@ -12,6 +12,7 @@ import type {ISnowflakeService} from '@app/api/infrastructure/ISnowflakeService'
 import type {IVoiceRoomStore} from '@app/api/infrastructure/IVoiceRoomStore';
 import type {UserCacheService} from '@app/api/infrastructure/UserCacheService';
 import type {RequestCache} from '@app/api/middleware/RequestCacheMiddleware';
+import type {Channel} from '@app/api/models/Channel';
 import type {ReadStateService} from '@app/api/read_state/ReadStateService';
 import type {IUserRepository} from '@app/api/user/IUserRepository';
 import type {VoiceAccessContext, VoiceAvailabilityService} from '@app/api/voice/VoiceAvailabilityService';
@@ -208,14 +209,26 @@ export class CallService {
 			has_reaction: false,
 			version: 1,
 		});
+		const author = await this.userRepository.findUnique(userId);
 		const call = await this.gatewayService.createCall(
 			channelId,
 			messageId.toString(),
 			selectedRegion,
 			ringing.map((id) => id.toString()),
 			allRecipients.map((id) => id.toString()),
+			author
+				? {
+						id: userId.toString(),
+						name: this.resolveCallerName({
+							channel,
+							userId,
+							globalName: author.globalName,
+							username: author.username,
+						}),
+						avatar: author.avatarHash,
+					}
+				: undefined,
 		);
-		const author = await this.userRepository.findUnique(userId);
 		await incrementDmMentionCounts({
 			readStateService: this.readStateService,
 			userRepository: this.userRepository,
@@ -390,13 +403,45 @@ export class CallService {
 				longitude,
 			});
 		} else {
+			const caller = await this.userCacheService.getUserPartialResponse(userId, requestCache);
 			await this.gatewayService.ringCallRecipients(
 				channelId,
 				recipientsToRing.map((id) => id.toString()),
+				{
+					id: userId.toString(),
+					name: this.resolveCallerName({
+						channel,
+						userId,
+						globalName: caller.global_name,
+						username: caller.username,
+					}),
+					avatar: caller.avatar,
+				},
 			);
 		}
 	}
 
+	private resolveCallerName({
+		channel,
+		userId,
+		globalName,
+		username,
+	}: {
+		channel: Channel;
+		userId: UserID;
+		globalName: string | null;
+		username: string;
+	}): string {
+		const nickname = channel.nicknames.get(userId.toString());
+		if (nickname) {
+			return nickname;
+		}
+		if (globalName) {
+			return globalName;
+		}
+		return username;
+	}
+
 	async stopRingingCallRecipients({
 		userId,
 		channelId,

fluxer_api/src/api/infrastructure/GatewayRpcClient.ts

@@ -3,7 +3,7 @@
 import {Config} from '@app/api/Config';
 import {GatewayRpcMethodError, GatewayRpcMethodErrorCodes} from '@app/api/infrastructure/GatewayRpcError';
 import type {IGatewayRpcTransport} from '@app/api/infrastructure/IGatewayRpcTransport';
-import type {CallData} from '@app/api/infrastructure/IGatewayService';
+import {type CallCaller, type CallData, callCallerRpcParams} from '@app/api/infrastructure/IGatewayService';
 import {NatsGatewayRpcTransport} from '@app/api/infrastructure/NatsGatewayRpcTransport';
 import {Logger} from '@app/api/Logger';
 import {NatsConnectionManager} from '@pkgs/nats/src/NatsConnectionManager';
@@ -128,6 +128,7 @@ export class GatewayRpcClient {
 		region: string,
 		ringing: Array<string>,
 		recipients: Array<string>,
+		caller?: CallCaller,
 	): Promise<CallData> {
 		return this.call<CallData>('call.create', {
 			channel_id: channelId,
@@ -135,6 +136,7 @@ export class GatewayRpcClient {
 			region,
 			ringing,
 			recipients,
+			...callCallerRpcParams(caller),
 		});
 	}
 
@@ -142,8 +144,8 @@ export class GatewayRpcClient {
 		return this.call('call.update_region', {channel_id: channelId, region});
 	}
 
-	async ringCallRecipients(channelId: string, recipients: Array<string>): Promise<boolean> {
-		return this.call('call.ring', {channel_id: channelId, recipients});
+	async ringCallRecipients(channelId: string, recipients: Array<string>, caller?: CallCaller): Promise<boolean> {
+		return this.call('call.ring', {channel_id: channelId, recipients, ...callCallerRpcParams(caller)});
 	}
 
 	async stopRingingCallRecipients(channelId: string, recipients: Array<string>): Promise<boolean> {

fluxer_api/src/api/infrastructure/GatewayService.ts

@@ -6,16 +6,18 @@ import {SYSTEM_USER_ID} from '@app/api/constants/Core';
 import type {GatewayDispatchEvent} from '@app/api/constants/Gateway';
 import {GatewayRpcClient} from '@app/api/infrastructure/GatewayRpcClient';
 import {GatewayRpcMethodError, GatewayRpcMethodErrorCodes} from '@app/api/infrastructure/GatewayRpcError';
-import type {
-	CallData,
-	GatewayChannelMention,
-	GatewayGuildMemoryStats,
-	GatewayMentionSources,
-	GatewayMentionSourcesPage,
-	GatewayNodeStats,
-	GatewayVoiceStateCounts,
-	GatewayVoiceStateEntry,
-	GuildChannelAuthContext,
+import {
+	type CallCaller,
+	type CallData,
+	callCallerRpcParams,
+	type GatewayChannelMention,
+	type GatewayGuildMemoryStats,
+	type GatewayMentionSources,
+	type GatewayMentionSourcesPage,
+	type GatewayNodeStats,
+	type GatewayVoiceStateCounts,
+	type GatewayVoiceStateEntry,
+	type GuildChannelAuthContext,
 } from '@app/api/infrastructure/IGatewayService';
 import {Logger} from '@app/api/Logger';
 import {APIErrorCodes} from '@fluxer/constants/src/ApiErrorCodes';
@@ -1695,6 +1697,7 @@ export class GatewayService {
 		region: string,
 		ringing: Array<string>,
 		recipients: Array<string>,
+		caller?: CallCaller,
 	): Promise<CallData> {
 		return this.call<CallData>('call.create', {
 			channel_id: channelId.toString(),
@@ -1702,6 +1705,7 @@ export class GatewayService {
 			region,
 			ringing,
 			recipients,
+			...callCallerRpcParams(caller),
 		});
 	}
 
@@ -1709,8 +1713,12 @@ export class GatewayService {
 		return this.call<boolean>('call.update_region', {channel_id: channelId.toString(), region});
 	}
 
-	async ringCallRecipients(channelId: ChannelID, recipients: Array<string>): Promise<boolean> {
-		return this.call<boolean>('call.ring', {channel_id: channelId.toString(), recipients});
+	async ringCallRecipients(channelId: ChannelID, recipients: Array<string>, caller?: CallCaller): Promise<boolean> {
+		return this.call<boolean>('call.ring', {
+			channel_id: channelId.toString(),
+			recipients,
+			...callCallerRpcParams(caller),
+		});
 	}
 
 	async stopRingingCallRecipients(channelId: ChannelID, recipients: Array<string>): Promise<boolean> {

fluxer_api/src/api/infrastructure/IGatewayService.ts

@@ -24,6 +24,19 @@ export interface CallData {
 	voice_states: Array<VoiceState>;
 }
 
+export interface CallCaller {
+	id: string;
+	name: string;
+	avatar: string | null;
+}
+
+export function callCallerRpcParams(caller: CallCaller | undefined): Record<string, unknown> {
+	if (!caller) {
+		return {};
+	}
+	return {caller_id: caller.id, caller_name: caller.name, caller_avatar: caller.avatar};
+}
+
 export interface GatewayGuildMemoryStatsEntry {
 	node_id: string;
 	guild_id: string | null;
@@ -381,11 +394,12 @@ export abstract class IGatewayService {
 		region: string,
 		ringing: Array<string>,
 		recipients: Array<string>,
+		caller?: CallCaller,
 	): Promise<CallData>;
 
 	abstract updateCallRegion(channelId: ChannelID, region: string | null): Promise<boolean>;
 
-	abstract ringCallRecipients(channelId: ChannelID, recipients: Array<string>): Promise<boolean>;
+	abstract ringCallRecipients(channelId: ChannelID, recipients: Array<string>, caller?: CallCaller): Promise<boolean>;
 
 	abstract stopRingingCallRecipients(channelId: ChannelID, recipients: Array<string>): Promise<boolean>;
 

fluxer_api/src/api/infrastructure/tests/CallRpcCallerParams.test.ts

@@ -0,0 +1,86 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {createChannelID} from '@app/api/BrandedTypes';
+import {GatewayRpcClient} from '@app/api/infrastructure/GatewayRpcClient';
+import {GatewayService} from '@app/api/infrastructure/GatewayService';
+import type {IGatewayRpcTransport} from '@app/api/infrastructure/IGatewayRpcTransport';
+import {afterEach, describe, expect, it} from 'vitest';
+
+const CHANNEL_ID = createChannelID(12n);
+
+interface RecordedCall {
+	method: string;
+	params: Record<string, unknown>;
+}
+
+function recordingService(recorded: Array<RecordedCall>): GatewayService {
+	const transport: IGatewayRpcTransport = {
+		async call(method: string, params: Record<string, unknown>): Promise<unknown> {
+			recorded.push({method, params});
+			return null;
+		},
+		async destroy(): Promise<void> {},
+	};
+	GatewayRpcClient.createForTests(transport);
+	return new GatewayService();
+}
+
+describe('call rpc caller params', () => {
+	afterEach(async () => {
+		await GatewayRpcClient.resetForTests();
+	});
+
+	it('sends the caller to call.create as caller_id, caller_name and caller_avatar', async () => {
+		const recorded: Array<RecordedCall> = [];
+		const service = recordingService(recorded);
+		await service.createCall(CHANNEL_ID, '99', 'automatic', ['2'], ['1', '2'], {
+			id: '1',
+			name: 'Elias',
+			avatar: 'a1b2c3d4',
+		});
+		expect(recorded).toHaveLength(1);
+		expect(recorded[0].method).toBe('call.create');
+		expect(recorded[0].params.caller_id).toBe('1');
+		expect(recorded[0].params.caller_name).toBe('Elias');
+		expect(recorded[0].params.caller_avatar).toBe('a1b2c3d4');
+	});
+
+	it('sends the caller to call.ring as caller_id, caller_name and caller_avatar', async () => {
+		const recorded: Array<RecordedCall> = [];
+		const service = recordingService(recorded);
+		await service.ringCallRecipients(CHANNEL_ID, ['2'], {id: '1', name: 'Elias', avatar: 'a1b2c3d4'});
+		expect(recorded).toHaveLength(1);
+		expect(recorded[0].method).toBe('call.ring');
+		expect(recorded[0].params.caller_id).toBe('1');
+		expect(recorded[0].params.caller_name).toBe('Elias');
+		expect(recorded[0].params.caller_avatar).toBe('a1b2c3d4');
+	});
+
+	it('sends caller_avatar as null when the caller has no avatar', async () => {
+		const recorded: Array<RecordedCall> = [];
+		const service = recordingService(recorded);
+		await service.ringCallRecipients(CHANNEL_ID, ['2'], {id: '1', name: 'Elias', avatar: null});
+		expect(recorded[0].params.caller_avatar).toBeNull();
+		expect(Object.hasOwn(recorded[0].params, 'caller_avatar')).toBe(true);
+	});
+
+	it('omits every caller key from call.create when no caller was resolved', async () => {
+		const recorded: Array<RecordedCall> = [];
+		const service = recordingService(recorded);
+		await service.createCall(CHANNEL_ID, '99', 'automatic', ['2'], ['1', '2']);
+		expect(recorded[0].params).toEqual({
+			channel_id: '12',
+			message_id: '99',
+			region: 'automatic',
+			ringing: ['2'],
+			recipients: ['1', '2'],
+		});
+	});
+
+	it('omits every caller key from call.ring when no caller was resolved', async () => {
+		const recorded: Array<RecordedCall> = [];
+		const service = recordingService(recorded);
+		await service.ringCallRecipients(CHANNEL_ID, ['2']);
+		expect(recorded[0].params).toEqual({channel_id: '12', recipients: ['2']});
+	});
+});

fluxer_api/src/api/test/NoopGatewayService.ts

@@ -19,6 +19,7 @@ import {GuildMemberRepository} from '@app/api/guild/repositories/GuildMemberRepo
 import {GuildRepository} from '@app/api/guild/repositories/GuildRepository';
 import {GuildRoleRepository} from '@app/api/guild/repositories/GuildRoleRepository';
 import {
+	type CallCaller,
 	type CallData,
 	type GatewayChannelMention,
 	type GatewayGuildMemoryStats,
@@ -925,6 +926,7 @@ export class NoopGatewayService extends IGatewayService {
 		_region: string,
 		_ringing: Array<string>,
 		_recipients: Array<string>,
+		_caller?: CallCaller,
 	): Promise<CallData> {
 		return {
 			channel_id: _channelId.toString(),
@@ -940,7 +942,7 @@ export class NoopGatewayService extends IGatewayService {
 		return true;
 	}
 
-	async ringCallRecipients(_channelId: ChannelID, _recipients: Array<string>): Promise<boolean> {
+	async ringCallRecipients(_channelId: ChannelID, _recipients: Array<string>, _caller?: CallCaller): Promise<boolean> {
 		return true;
 	}
 

fluxer_docs/src/content/docs/http-api/users/push-notifications.mdx

@@ -10,16 +10,16 @@ Fluxer delivers a notification to a registered device as [RFC 8291](https://data
 
 Every route here requires a user session. A bot or OAuth2 bearer credential is refused with 403 `ACCESS_DENIED`. An account with an outstanding required action is refused with 403 `ACCOUNT_SUSPICIOUS_ACTIVITY`.
 
-## Registration shapes
+## Registration formats
 
-A registration takes one of two shapes on every platform.
+On every platform a registration takes one of two forms.
 
-| Shape | What `token` holds | Keys |
+| Form | What `token` holds | Keys |
 | --- | --- | --- |
 | Web Push | A publicly routable endpoint URL | Both `encryption_key` and `auth_secret` |
 | Legacy | A raw vendor device token | Neither key is sent |
 
-Fluxer reads the shape from the body rather than from `platform`. A `token` that parses as a URL without both keys is refused, and so is a pair of keys sent with a raw vendor token.
+Fluxer works out which form you sent from the body rather than from `platform`. A `token` that parses as a URL without both keys is refused, and so is a pair of keys sent with a raw vendor token.
 
 `platform` names the transport the device was reached on.
 
@@ -34,6 +34,8 @@ Fluxer reads the shape from the body rather than from `platform`. A `token` that
 
 Apple issues a PushKit device token separate from the alert token. An iOS device that answers calls registers both: the alert token as `ios_apns`, and the PushKit token as `ios_apns_voip`. Generate a separate key pair for the `ios_apns_voip` registration. The two registrations get two identifiers and two independent lifetimes. Removing one leaves the other in place.
 
+Android registers once. One `android_fcm` Web Push registration receives notifications, clears and call rings. Android has no separate call registration and no second token. An `android_fcm` registration sent with neither key receives no call ring.
+
 ## Device registration object
 
 The identifier Fluxer assigns to one registration.
@@ -56,7 +58,7 @@ Stores a push registration for the current account and returns its [device regis
 
 | Field | Type | Description |
 | --- | --- | --- |
-| platform | string | The [platform value](#registration-shapes) the device was reached on |
+| platform | string | The [platform value](#registration-formats) the device was reached on |
 | token<sup>1</sup> | string | The endpoint URL, or the raw vendor token on a legacy registration |
 | encryption_key?<sup>2</sup> | string | The base64url P-256 public key (1-1024 characters) |
 | auth_secret?<sup>2</sup> | string | The base64url auth secret (1-1024 characters) |
@@ -77,7 +79,7 @@ Register an `https` endpoint. A Web Push registration whose `token` is not a val
 | Status | Body | Condition |
 | --- | --- | --- |
 | 200 | [device registration](#device-registration-object) object | The registration was stored |
-| 400 | [error response](/http-api/#error-response) | The body matches neither shape and the request returns `INVALID_FORM_BODY` |
+| 400 | [error response](/http-api/#error-response) | The body matches neither form and the request returns `INVALID_FORM_BODY` |
 
 ### Rate limit
 
@@ -95,7 +97,7 @@ The four values below identify the registration the same way [Register mobile pu
 
 | Field | Type | Description |
 | --- | --- | --- |
-| platform | string | The [platform value](#registration-shapes) sent at registration |
+| platform | string | The [platform value](#registration-formats) sent at registration |
 | token | string | The endpoint URL or raw vendor token sent at registration (1-4096 characters) |
 | app_id? | string | The client build sent at registration (default `stable`) |
 | provider_environment? | string | The environment sent at registration |
@@ -148,7 +150,9 @@ An endpoint that answers 404 or 410 removes the registration. Fluxer retries a t
 
 ### Call ring
 
-A call ring is the only payload an `ios_apns_voip` registration receives. Every other payload for that device goes to its `ios_apns` registration. The fields below sit under `data`.
+Only two kinds of registration receive a call ring. On iOS it goes to the `ios_apns_voip` registration, which receives nothing else, while every other payload for that device goes to its `ios_apns` registration. An `android_fcm` Web Push registration receives call rings alongside its notifications and clears. No other registration receives one.
+
+These fields are under `data`.
 
 | Field | Type | Description |
 | --- | --- | --- |
@@ -157,14 +161,34 @@ A call ring is the only payload an `ios_apns_voip` registration receives. Every
 | message_id | string | The call message, which names the call |
 | target_user_id | string | The account being rung |
 | started_at_ms | integer | When the ring started, in milliseconds since the Unix epoch |
+| expires_at_ms | integer | When the ring window closes, in milliseconds since the Unix epoch |
+| caller_id? | string | The account placing the call |
+| caller_name? | string | The name to show for the caller, resolved before the record was sealed |
+| caller_avatar_url? | string | An image URL for the caller |
+
+Stop ringing at `expires_at_ms` rather than after a fixed number of seconds. The ring window can change without a client release.
+
+`caller_name` is the caller's nickname in that channel if they have one, otherwise their display name, otherwise their username. Display it as sent. If you look the name up yourself from `caller_id` you can get a different answer in a group DM.
+
+The three caller fields are sent together or not at all. An older server, or an account Fluxer could not read, omits all three.
+
+| What arrived | What to show |
+| --- | --- |
+| The three caller fields | The name and the image |
+| A name with no image URL | The name and a local placeholder |
+| No caller fields | The channel name the client already holds, or a generic incoming call string |
+
+Ring in all three cases. Showing a call without a name is better than not showing it at all.
 
 A call ring is not stored for later delivery. A device that cannot be reached while the call rings does not get the ring afterwards.
 
-Nothing cancels a ring with a second push. Fluxer ends the call over the gateway connection the woken client opens.
+Fluxer never sends a second push to cancel a ring. It ends the call over the gateway connection that the woken client opens.
+
+A call also produces an ordinary notification for its call message, under the same `message_id` as the ring. Collapse the two into one rather than showing both.
 
 ### What PushKit requires of the client
 
-iOS terminates an application that takes a PushKit push without reporting a call to CallKit. Repeated failures stop PushKit delivery to that device. The report is due before the record is decrypted. Decryption cannot be what decides whether to ring.
+iOS terminates an application that takes a PushKit push without reporting a call to CallKit. Repeated failures stop PushKit delivery to that device. You have to report the call before you decrypt the record, so decryption cannot decide whether to ring.
 
 Report a call for every PushKit push, before decrypting. Then end that call at once in each of these three cases.
 
@@ -174,11 +198,43 @@ Report a call for every PushKit push, before decrypting. Then end that call at o
 | `type` is not `call_ring` | The push did not come from Fluxer |
 | The gateway names no live call for `channel_id` | The call ended before the ring arrived |
 
-Anyone who learns a PushKit token can send to it. Those three rules are what keeps a forged push from showing a caller.
+Anyone who obtains a PushKit token can send to it. These three checks are what stop a forged push from showing a caller.
 
-Derive the CallKit call identifier from `message_id`. The gateway ends the call over the connection under that same identity. Two rings for one call then name one call.
+Derive the CallKit call identifier from `message_id`. The gateway uses the same identifier when it ends the call, and two rings for the same call resolve to a single CallKit call.
 
-Keep the registered private key and auth secret readable while the device is locked. A call ring arrives on a locked device. A key that cannot be read then costs the report.
+Keep the registered private key and auth secret readable while the device is locked, because call rings arrive on locked devices. If you cannot read the key you cannot decrypt the record, but CallKit still needs its report.
+
+### What an Android client has to do
+
+Android has no PushKit, so there is nothing to do before decrypting. Decrypt the record, read `type`, then decide whether to ring. The CallKit rule above applies only to iOS.
+
+A ring arrives as a high priority data message and is never queued. Your own process handles it and has the registered private key, so decrypt it there. Do not make any network request before you have posted a notification.
+
+Google lowers an app's message priority if its high priority messages stop producing a visible notification. This happens per device and can be reversed. Once downgraded, a ring waits for the device to leave Doze and arrives after the call has already ended, so every ring needs to put something on screen.
+
+1. Post the incoming call notification first, ahead of any step that can fail. Use `Notification.CallStyle.forIncomingCall` from Android 12 on a high importance channel, with `CATEGORY_CALL`, a full screen intent, `setOngoing(true)`, and a timeout taken from `expires_at_ms`. Below Android 12 post a high importance notification with the same full screen intent.
+2. Post with `caller_name` alone, then fetch `caller_avatar_url`, then post again under the same notification id with the image attached. Read the active notifications before that second post. Without that read a ring the user already dismissed comes back.
+3. Show a fallback when the record does not decrypt, or when `type` is not `call_ring`. Returning without showing anything is what triggers the downgrade described above.
+4. Let the system play the ringtone through the notification channel. A ringtone the app starts itself out of a push stops working on newer Android releases and reports no error.
+5. Stop registering the device once notification permission is denied. An app that cannot show anything is spending its delivery priority for no reason.
+
+Answering the call needs a foreground service. Ringing does not. Start a `phoneCall` service once the user answers. From Android 12 a backgrounded app may only start that service because the message arrived at high priority. Read the delivered priority rather than the priority Fluxer sent, and skip the service start when it came down as normal.
+
+| Permission | Granted by default | Without it |
+| --- | --- | --- |
+| POST_NOTIFICATIONS | No, the user is asked from Android 13 | Nothing renders at all |
+| USE_FULL_SCREEN_INTENT | Yes for a calling app, checked at runtime from Android 14 | The ring shows as a heads-up notification |
+| MANAGE_OWN_CALLS | Yes, at install | The `phoneCall` service type is unusable |
+
+Google Play grants `USE_FULL_SCREEN_INTENT` to an app whose stated function includes calling and revokes it for everything else. Read `NotificationManager.canUseFullScreenIntent()` at runtime and send the user to the system setting when it answers false. Losing it does not lose the ring.
+
+In three situations nothing is delivered at all. These are worth documenting for support rather than treating as bugs.
+
+| Situation | What happens |
+| --- | --- |
+| The user force stopped the app | Nothing is delivered until the app is opened again |
+| The app is restricted from background work | Nothing is delivered |
+| The device has not been unlocked since it booted | The key is unreadable and the ring cannot be decrypted |
 
 ### When decryption fails
 

fluxer_gateway/src/call/call.erl

@@ -12,7 +12,10 @@
     message_id := integer(),
     region := binary() | undefined,
     ringing := [integer()],
-    recipients := [integer()]
+    recipients := [integer()],
+    caller_id => integer() | undefined,
+    caller_name => binary() | undefined,
+    caller_avatar => binary() | undefined
 }.
 -type call_request() ::
     {get_state}
@@ -28,7 +31,9 @@
     | {update_voice_state, integer(), map()}
     | {get_sessions}
     | {get_pending_connections}.
--type cast_request() :: {join_async, integer(), map(), binary(), pid()}.
+-type cast_request() ::
+    {join_async, integer(), map(), binary(), pid()}
+    | {set_caller, map()}.
 -type info_message() ::
     {'DOWN', reference(), process, pid(), term()}
     | {ring_timeout, integer()}
@@ -58,20 +63,25 @@ init(CallData) ->
     {ok, FinalState}.
 
 -spec build_initial_state(call_data()) -> map().
-build_initial_state(#{
-    channel_id := ChannelId,
-    message_id := MessageId,
-    region := Region,
-    ringing := Ringing,
-    recipients := Recipients
-}) ->
+build_initial_state(
     #{
+        channel_id := ChannelId,
+        message_id := MessageId,
+        region := Region,
+        ringing := Ringing,
+        recipients := Recipients
+    } = CallData
+) ->
+    State = #{
         channel_id => ChannelId,
         message_id => MessageId,
         region => Region,
         ringing => [],
         pending_ringing => Ringing,
         recipients => Recipients,
+        caller_id => undefined,
+        caller_name => undefined,
+        caller_avatar => undefined,
         voice_states => #{},
         sessions => #{},
         pending_connections => #{},
@@ -81,7 +91,8 @@ build_initial_state(#{
         created_at => erlang:system_time(millisecond),
         participants_history => sets:new(),
         last_call_event => undefined
-    }.
+    },
+    call_state:put_caller(CallData, State).
 
 -spec run_init_pipeline(map()) -> map().
 run_init_pipeline(State) ->
@@ -155,6 +166,8 @@ handle_cast(Request, State) ->
     case decode_cast_request(Request) of
         {ok, {join_async, UserId, VoiceState, SessionId, SessionPid}} ->
             call_voice:handle_join_async(UserId, VoiceState, SessionId, SessionPid, State);
+        {ok, {set_caller, Caller}} ->
+            {noreply, call_state:put_caller(Caller, State)};
         error ->
             {noreply, State}
     end.
@@ -308,6 +321,8 @@ decode_cast_request({join_async, UserId, VoiceState, SessionId, SessionPid}) whe
     is_integer(UserId), is_map(VoiceState), is_binary(SessionId), is_pid(SessionPid)
 ->
     {ok, {join_async, UserId, VoiceState, SessionId, SessionPid}};
+decode_cast_request({set_caller, Caller}) when is_map(Caller) ->
+    {ok, {set_caller, Caller}};
 decode_cast_request(_) ->
     error.
 

fluxer_gateway/src/call/call_handoff.erl

@@ -16,6 +16,9 @@ export_state(State) ->
         ringing => maps:get(ringing, State, []),
         pending_ringing => maps:get(pending_ringing, State, []),
         recipients => maps:get(recipients, State, []),
+        caller_id => maps:get(caller_id, State, undefined),
+        caller_name => maps:get(caller_name, State, undefined),
+        caller_avatar => maps:get(caller_avatar, State, undefined),
         voice_states => maps:get(voice_states, State, #{}),
         sessions => export_sessions(maps:get(sessions, State, #{})),
         pending_connections => maps:get(pending_connections, State, #{}),
@@ -40,6 +43,9 @@ restore_state(TransferState) ->
         ringing => maps:get(ringing, TransferState, []),
         pending_ringing => maps:get(pending_ringing, TransferState, []),
         recipients => maps:get(recipients, TransferState, []),
+        caller_id => maps:get(caller_id, TransferState, undefined),
+        caller_name => maps:get(caller_name, TransferState, undefined),
+        caller_avatar => maps:get(caller_avatar, TransferState, undefined),
         voice_states => VoiceStates,
         sessions => Sessions,
         pending_connections => maps:get(pending_connections, TransferState, #{}),
@@ -192,4 +198,25 @@ restore_state_remonitors_live_sessions_test() ->
     ?assertEqual(#{1 => VoiceState}, maps:get(voice_states, Restored)),
     SessionPid ! stop.
 
+export_and_restore_round_trips_the_caller_test() ->
+    State = #{
+        channel_id => 123,
+        message_id => 456,
+        caller_id => 7,
+        caller_name => <<"Ada">>,
+        caller_avatar => <<"a1b2c3d4">>
+    },
+    Restored = restore_state(export_state(State)),
+    ?assertEqual(
+        #{caller_id => 7, caller_name => <<"Ada">>, caller_avatar => <<"a1b2c3d4">>},
+        call_state:caller_from_state(Restored)
+    ).
+
+restore_state_reads_a_transfer_without_a_caller_as_undefined_test() ->
+    Restored = restore_state(#{channel_id => 123, message_id => 456}),
+    ?assertEqual(
+        #{caller_id => undefined, caller_name => undefined, caller_avatar => undefined},
+        call_state:caller_from_state(Restored)
+    ).
+
 -endif.

fluxer_gateway/src/call/call_ringing.erl

@@ -89,41 +89,51 @@ apply_pending_ringing(PendingUnique, State, DispatchUpdates) ->
 publish_rings([], _State) ->
     ok;
 publish_rings(UserIds, State) ->
-    ChannelId = maps:get(channel_id, State),
-    MessageId = maps:get(message_id, State),
     Recipients = maps:get(recipients, State),
-    StartedAt = erlang:system_time(millisecond),
-    ExpiresAt = StartedAt + ?RING_TIMEOUT_MS,
-    _ = proc_lib:spawn(fun() ->
-        publish_ring_jobs(UserIds, Recipients, ChannelId, MessageId, StartedAt, ExpiresAt)
-    end),
+    Ring = build_ring_context(State),
+    _ = proc_lib:spawn(fun() -> publish_ring_jobs(UserIds, Recipients, Ring) end),
     ok.
 
--spec publish_ring_jobs(
-    [integer()], [integer()], integer(), integer(), integer(), integer()
-) -> ok.
-publish_ring_jobs(UserIds, Recipients, ChannelId, MessageId, StartedAt, ExpiresAt) ->
+-spec build_ring_context(map()) -> map().
+build_ring_context(State) ->
+    StartedAt = erlang:system_time(millisecond),
+    #{
+        channel_id => maps:get(channel_id, State),
+        message_id => maps:get(message_id, State),
+        started_at_ms => StartedAt,
+        expires_at_ms => StartedAt + ?RING_TIMEOUT_MS,
+        caller => call_state:caller_from_state(State)
+    }.
+
+-spec publish_ring_jobs([integer()], [integer()], map()) -> ok.
+publish_ring_jobs(UserIds, Recipients, Ring) ->
     lists:foreach(
-        fun(UserId) ->
-            publish_ring_job(UserId, Recipients, ChannelId, MessageId, StartedAt, ExpiresAt)
-        end,
+        fun(UserId) -> publish_ring_job(UserId, Recipients, Ring) end,
         UserIds
     ).
 
--spec publish_ring_job(
-    integer(), [integer()], integer(), integer(), integer(), integer()
-) -> ok.
-publish_ring_job(UserId, Recipients, ChannelId, MessageId, StartedAt, ExpiresAt) ->
+-spec publish_ring_job(integer(), [integer()], map()) -> ok.
+publish_ring_job(UserId, Recipients, Ring) ->
     case ring_suppressed(UserId, Recipients) of
         true ->
             ok;
         false ->
-            _ = push_job_publisher:publish_ring(
-                UserId, ChannelId, MessageId, StartedAt, ExpiresAt
-            ),
-            ok
+            dispatch_ring_job(UserId, Ring)
     end.
 
+-spec dispatch_ring_job(integer(), map()) -> ok.
+dispatch_ring_job(UserId, #{
+    channel_id := ChannelId,
+    message_id := MessageId,
+    started_at_ms := StartedAt,
+    expires_at_ms := ExpiresAt,
+    caller := Caller
+}) ->
+    _ = push_job_publisher:publish_ring(
+        UserId, ChannelId, MessageId, StartedAt, ExpiresAt, Caller
+    ),
+    ok.
+
 -spec ring_suppressed(integer(), [integer()]) -> boolean().
 ring_suppressed(UserId, Recipients) ->
     all_blocked(UserId, [Other || Other <- Recipients, Other =/= UserId]).
@@ -382,11 +392,14 @@ published_ring_args() ->
     [Args || {_Pid, {push_job_publisher, publish_ring, Args}, _Result} <- History].
 
 published_ring_users() ->
-    lists:usort([UserId || [UserId, _C, _M, _S, _E] <- published_ring_args()]).
+    lists:usort([UserId || [UserId, _C, _M, _S, _E, _Caller] <- published_ring_args()]).
+
+published_ring_callers() ->
+    [Caller || [_U, _C, _M, _S, _E, Caller] <- published_ring_args()].
 
 mock_ring_publisher() ->
     ok = meck:new(push_job_publisher, [passthrough, no_link]),
-    meck:expect(push_job_publisher, publish_ring, fun(_U, _C, _M, _S, _E) -> ok end).
+    meck:expect(push_job_publisher, publish_ring, fun(_U, _C, _M, _S, _E, _Caller) -> ok end).
 
 apply_pending_ringing_publishes_one_ring_per_new_user_test() ->
     ok = mock_ring_publisher(),
@@ -405,7 +418,7 @@ apply_pending_ringing_publishes_the_frozen_ring_fields_test() ->
         State = ring_state([1, 2], [], [2], #{}),
         _ = maybe_dispatch_pending_ringing(State, false),
         ok = meck:wait(1, push_job_publisher, publish_ring, '_', 2000),
-        [[UserId, ChannelId, MessageId, StartedAt, ExpiresAt]] = published_ring_args(),
+        [[UserId, ChannelId, MessageId, StartedAt, ExpiresAt, _Caller]] = published_ring_args(),
         ?assertEqual({2, 100, 200}, {UserId, ChannelId, MessageId}),
         ?assertEqual(?RING_TIMEOUT_MS, ExpiresAt - StartedAt)
     after
@@ -447,13 +460,53 @@ publish_ring_jobs_suppresses_a_fully_blocked_ring_test() ->
     ok = meck:new(push_eligibility, [passthrough, no_link]),
     try
         ok = meck:expect(push_eligibility, is_user_blocked, fun(_U, _O) -> true end),
-        ok = publish_ring_jobs([2], [1, 2], 100, 200, 1, 2),
+        ok = publish_ring_jobs(
+            [2], [1, 2], build_ring_context(ring_state([1, 2], [], [], #{}))
+        ),
         ?assertEqual([], published_ring_args())
     after
         meck:unload(push_eligibility),
         meck:unload(push_job_publisher)
     end.
 
+apply_pending_ringing_publishes_the_caller_held_on_state_test() ->
+    ok = mock_ring_publisher(),
+    try
+        State = (ring_state([1, 2], [], [2], #{}))#{
+            caller_id => 1,
+            caller_name => <<"Ada">>,
+            caller_avatar => <<"a1b2c3d4">>
+        },
+        _ = maybe_dispatch_pending_ringing(State, false),
+        ok = meck:wait(1, push_job_publisher, publish_ring, '_', 2000),
+        ?assertEqual(
+            [
+                #{
+                    caller_id => 1,
+                    caller_name => <<"Ada">>,
+                    caller_avatar => <<"a1b2c3d4">>
+                }
+            ],
+            published_ring_callers()
+        )
+    after
+        meck:unload(push_job_publisher)
+    end.
+
+apply_pending_ringing_publishes_an_unresolved_caller_when_state_has_none_test() ->
+    ok = mock_ring_publisher(),
+    try
+        State = ring_state([1, 2], [], [2], #{}),
+        _ = maybe_dispatch_pending_ringing(State, false),
+        ok = meck:wait(1, push_job_publisher, publish_ring, '_', 2000),
+        ?assertEqual(
+            [#{caller_id => undefined, caller_name => undefined, caller_avatar => undefined}],
+            published_ring_callers()
+        )
+    after
+        meck:unload(push_job_publisher)
+    end.
+
 ring_is_suppressed_when_every_other_recipient_is_blocked_test() ->
     ok = meck:new(push_eligibility, [passthrough, no_link]),
     try

fluxer_gateway/src/call/call_state.erl

@@ -5,6 +5,8 @@
 
 -export([
     build_call_event/1,
+    caller_from_state/1,
+    put_caller/2,
     format_voice_state/1,
     format_voice_states/1,
     format_pending_connections/1,
@@ -28,6 +30,26 @@ build_call_event(State) ->
         voice_states => format_voice_states(maps:get(voice_states, State))
     }.
 
+-spec caller_from_state(map()) -> map().
+caller_from_state(State) ->
+    #{
+        caller_id => maps:get(caller_id, State, undefined),
+        caller_name => maps:get(caller_name, State, undefined),
+        caller_avatar => maps:get(caller_avatar, State, undefined)
+    }.
+
+-spec put_caller(map(), map()) -> map().
+put_caller(#{caller_id := Id, caller_name := Name} = Caller, State) when
+    is_integer(Id), is_binary(Name), byte_size(Name) > 0
+->
+    State#{
+        caller_id => Id,
+        caller_name => Name,
+        caller_avatar => maps:get(caller_avatar, Caller, undefined)
+    };
+put_caller(_Caller, State) ->
+    State.
+
 -spec integer_list_to_binaries([integer()]) -> [binary()].
 integer_list_to_binaries(Values) ->
     lists:map(fun integer_to_binary/1, Values).
@@ -245,3 +267,39 @@ voice_state_connection_id(VoiceState) ->
         VoiceState,
         maps:get(connection_id, VoiceState, undefined)
     ).
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
+put_caller_stores_a_resolved_caller_test() ->
+    Caller = #{caller_id => 7, caller_name => <<"Ada">>, caller_avatar => <<"a1b2c3d4">>},
+    ?assertEqual(Caller, caller_from_state(put_caller(Caller, #{}))).
+
+put_caller_overwrites_an_earlier_caller_test() ->
+    First = #{caller_id => 7, caller_name => <<"Ada">>, caller_avatar => <<"a1b2c3d4">>},
+    Second = #{caller_id => 9, caller_name => <<"Bo">>, caller_avatar => undefined},
+    State = put_caller(Second, put_caller(First, #{})),
+    ?assertEqual(Second, caller_from_state(State)).
+
+put_caller_keeps_the_stored_caller_when_the_new_one_is_unresolved_test() ->
+    Caller = #{caller_id => 7, caller_name => <<"Ada">>, caller_avatar => <<"a1b2c3d4">>},
+    State = put_caller(Caller, #{}),
+    Unresolved = #{
+        caller_id => undefined, caller_name => undefined, caller_avatar => undefined
+    },
+    ?assertEqual(Caller, caller_from_state(put_caller(Unresolved, State))).
+
+put_caller_rejects_a_caller_without_a_name_test() ->
+    Caller = #{caller_id => 7, caller_name => <<>>, caller_avatar => undefined},
+    ?assertEqual(
+        #{caller_id => undefined, caller_name => undefined, caller_avatar => undefined},
+        caller_from_state(put_caller(Caller, #{}))
+    ).
+
+caller_from_state_reads_an_absent_caller_as_undefined_test() ->
+    ?assertEqual(
+        #{caller_id => undefined, caller_name => undefined, caller_avatar => undefined},
+        caller_from_state(#{})
+    ).
+
+-endif.

fluxer_gateway/src/gateway/gateway_rpc_call.erl

@@ -19,10 +19,11 @@ execute_method(<<"call.update_region">>, #{
     <<"channel_id">> := ChannelIdBin, <<"region">> := Region
 }) ->
     handle_call_update_region(ChannelIdBin, Region);
-execute_method(<<"call.ring">>, #{
-    <<"channel_id">> := ChannelIdBin, <<"recipients">> := RecipientsBin
-}) ->
-    handle_call_ring(ChannelIdBin, RecipientsBin);
+execute_method(
+    <<"call.ring">>,
+    #{<<"channel_id">> := ChannelIdBin, <<"recipients">> := RecipientsBin} = Params
+) ->
+    handle_call_ring(ChannelIdBin, RecipientsBin, Params);
 execute_method(<<"call.stop_ringing">>, #{
     <<"channel_id">> := ChannelIdBin, <<"recipients">> := RecipientsBin
 }) ->
@@ -117,15 +118,32 @@ handle_call_create(Params) ->
     MessageId = validation:snowflake_or_throw(<<"message_id">>, MessageIdBin),
     Ringing = validation:snowflake_list_or_throw(<<"ringing">>, RingingBins),
     Recipients = validation:snowflake_list_or_throw(<<"recipients">>, RecipientsBins),
-    CallData = #{
-        channel_id => ChannelId,
-        message_id => MessageId,
-        region => Region,
-        ringing => Ringing,
-        recipients => Recipients
-    },
+    CallData = maps:merge(
+        #{
+            channel_id => ChannelId,
+            message_id => MessageId,
+            region => Region,
+            ringing => Ringing,
+            recipients => Recipients
+        },
+        caller_params(Params)
+    ),
     do_call_create(ChannelId, CallData).
 
+-spec caller_params(map()) -> map().
+caller_params(Params) ->
+    #{
+        caller_id => snowflake_id:parse_maybe(maps:get(<<"caller_id">>, Params, undefined)),
+        caller_name => optional_binary(maps:get(<<"caller_name">>, Params, undefined)),
+        caller_avatar => optional_binary(maps:get(<<"caller_avatar">>, Params, undefined))
+    }.
+
+-spec optional_binary(term()) -> binary() | undefined.
+optional_binary(Value) when is_binary(Value), byte_size(Value) > 0 ->
+    Value;
+optional_binary(_Value) ->
+    undefined.
+
 -spec handle_call_update_region(binary(), binary()) -> term().
 handle_call_update_region(ChannelIdBin, Region) ->
     ChannelId = validation:snowflake_or_throw(<<"channel_id">>, ChannelIdBin),
@@ -142,16 +160,18 @@ update_region(Pid, Region) ->
         _ -> gateway_rpc_error:raise(<<"update_region_error">>)
     end.
 
--spec handle_call_ring(binary(), list()) -> term().
-handle_call_ring(ChannelIdBin, RecipientsBin) ->
+-spec handle_call_ring(binary(), list(), map()) -> term().
+handle_call_ring(ChannelIdBin, RecipientsBin, Params) ->
     ChannelId = validation:snowflake_or_throw(<<"channel_id">>, ChannelIdBin),
     Recipients = validation:snowflake_list_or_throw(<<"recipients">>, RecipientsBin),
+    Caller = caller_params(Params),
     gateway_rpc_call_lookup:with_call(ChannelId, fun(Pid) ->
-        ring_recipients(Pid, Recipients)
+        ring_recipients(Pid, Recipients, Caller)
     end).
 
--spec ring_recipients(pid(), [integer()]) -> true.
-ring_recipients(Pid, Recipients) ->
+-spec ring_recipients(pid(), [integer()], map()) -> true.
+ring_recipients(Pid, Recipients, Caller) ->
+    ok = set_call_caller(Pid, Caller),
     case
         gateway_rpc_call_lookup:safe_gen_server_call(
             Pid, {ring_recipients, Recipients}, ?CALL_LOOKUP_TIMEOUT
@@ -161,6 +181,13 @@ ring_recipients(Pid, Recipients) ->
         _ -> gateway_rpc_error:raise(<<"ring_recipients_error">>)
     end.
 
+-spec set_call_caller(pid(), map()) -> ok.
+set_call_caller(_Pid, #{caller_id := undefined}) ->
+    ok;
+set_call_caller(Pid, Caller) ->
+    gen_server:cast(Pid, {set_caller, Caller}),
+    ok.
+
 -spec handle_call_stop_ringing(binary(), list()) -> term().
 handle_call_stop_ringing(ChannelIdBin, RecipientsBin) ->
     ChannelId = validation:snowflake_or_throw(<<"channel_id">>, ChannelIdBin),

fluxer_gateway/src/push/push_job_publisher.erl

@@ -4,13 +4,14 @@
 -typing([eqwalizer]).
 
 -export([publish_message/8, publish_message/10, publish_clear/3, publish_clear/5]).
--export([publish_ring/5, request/3]).
+-export([publish_ring/6, request/3]).
 
 -define(SUBJECT_MESSAGE, <<"push.job.message">>).
 -define(SUBJECT_CLEAR, <<"push.job.clear">>).
 -define(SUBJECT_RING, <<"push.job.ring">>).
 -define(JOB_VERSION, 1).
 -define(NATS_MAX_PAYLOAD_BYTES, 1048576).
+-define(MAX_CALLER_NAME_BYTES, 128).
 
 -type meta() :: #{
     kind := message | clear | ring,
@@ -134,18 +135,21 @@ publish_clear(UserId, ChannelId, MessageId, ConfigVersion, Fallback) ->
         fallback => Fallback
     }).
 
--spec publish_ring(integer(), integer(), integer(), integer(), integer()) ->
+-spec publish_ring(integer(), integer(), integer(), integer(), integer(), map()) ->
     ok | {error, term()}.
-publish_ring(UserId, ChannelId, MessageId, StartedAtMs, ExpiresAtMs) ->
-    Job = #{
-        <<"v">> => ?JOB_VERSION,
-        <<"config_version">> => push_delivery_config:config_version(),
-        <<"user_id">> => integer_to_binary(UserId),
-        <<"channel_id">> => integer_to_binary(ChannelId),
-        <<"message_id">> => integer_to_binary(MessageId),
-        <<"started_at_ms">> => StartedAtMs,
-        <<"expires_at_ms">> => ExpiresAtMs
-    },
+publish_ring(UserId, ChannelId, MessageId, StartedAtMs, ExpiresAtMs, Caller) ->
+    Job = maps:merge(
+        #{
+            <<"v">> => ?JOB_VERSION,
+            <<"config_version">> => push_delivery_config:config_version(),
+            <<"user_id">> => integer_to_binary(UserId),
+            <<"channel_id">> => integer_to_binary(ChannelId),
+            <<"message_id">> => integer_to_binary(MessageId),
+            <<"started_at_ms">> => StartedAtMs,
+            <<"expires_at_ms">> => ExpiresAtMs
+        },
+        caller_fields(Caller)
+    ),
     publish(?SUBJECT_RING, Job, #{
         kind => ring,
         user_ids => [UserId],
@@ -154,6 +158,33 @@ publish_ring(UserId, ChannelId, MessageId, StartedAtMs, ExpiresAtMs) ->
         fallback => fun ignore_fallback/1
     }).
 
+-spec caller_fields(map()) -> map().
+caller_fields(#{caller_id := CallerId, caller_name := Name} = Caller) when
+    is_integer(CallerId), is_binary(Name), byte_size(Name) > 0
+->
+    CallerIdBin = integer_to_binary(CallerId),
+    #{
+        <<"caller_id">> => CallerIdBin,
+        <<"caller_name">> => push_notification_format:truncate_bytes(
+            Name, ?MAX_CALLER_NAME_BYTES
+        ),
+        <<"caller_avatar_url">> => caller_avatar_url(
+            CallerIdBin, maps:get(caller_avatar, Caller, undefined)
+        )
+    };
+caller_fields(_Caller) ->
+    #{}.
+
+-spec caller_avatar_url(binary(), term()) -> binary().
+caller_avatar_url(CallerIdBin, Hash) when is_binary(Hash), byte_size(Hash) > 0 ->
+    push_notification_format:resolve_author_avatar_url(#{
+        <<"id">> => CallerIdBin, <<"avatar">> => Hash
+    });
+caller_avatar_url(CallerIdBin, _Hash) ->
+    push_notification_format:resolve_author_avatar_url(#{
+        <<"id">> => CallerIdBin, <<"avatar">> => null
+    }).
+
 -spec request(binary(), binary(), pos_integer()) -> ok | {error, term()}.
 request(Subject, Body, Timeout) ->
     case gateway_nats_pool_conn:get_pool_conn() of
@@ -269,3 +300,77 @@ outbox_job(Subject, Job, Body, Meta) ->
         message_id => MessageId,
         fallback => Fallback
     }.
+
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+
+unresolved_caller() ->
+    #{caller_id => undefined, caller_name => undefined, caller_avatar => undefined}.
+
+with_endpoint_env(Fun) ->
+    ok = meck:new(fluxer_gateway_env, [passthrough, no_link]),
+    try
+        ok = meck:expect(fluxer_gateway_env, get, fun endpoint_env_meck/1),
+        Fun()
+    after
+        meck:unload(fluxer_gateway_env)
+    end.
+
+endpoint_env_meck(media_proxy_endpoint) -> <<"https://media.example">>;
+endpoint_env_meck(static_cdn_endpoint) -> <<"https://static.example">>;
+endpoint_env_meck(Key) -> meck:passthrough([Key]).
+
+caller_fields_omits_every_key_when_the_caller_is_unresolved_test() ->
+    ?assertEqual(#{}, caller_fields(unresolved_caller())).
+
+caller_fields_omits_every_key_when_only_the_name_resolved_test() ->
+    ?assertEqual(
+        #{},
+        caller_fields(#{
+            caller_id => undefined, caller_name => <<"Ada">>, caller_avatar => undefined
+        })
+    ).
+
+caller_fields_builds_the_avatar_url_from_the_hash_test() ->
+    Fields = with_endpoint_env(fun() ->
+        caller_fields(#{
+            caller_id => 1234567890123456789,
+            caller_name => <<"Ada">>,
+            caller_avatar => <<"a1b2c3d4">>
+        })
+    end),
+    ?assertEqual(
+        #{
+            <<"caller_id">> => <<"1234567890123456789">>,
+            <<"caller_name">> => <<"Ada">>,
+            <<"caller_avatar_url">> =>
+                <<"https://media.example/avatars/1234567890123456789/a1b2c3d4.png">>
+        },
+        Fields
+    ).
+
+caller_fields_falls_back_to_the_default_avatar_test() ->
+    Fields = with_endpoint_env(fun() ->
+        caller_fields(#{
+            caller_id => 1234567890123456789,
+            caller_name => <<"Ada">>,
+            caller_avatar => undefined
+        })
+    end),
+    ?assertMatch(
+        #{<<"caller_avatar_url">> := <<"https://static.example/avatars/", _/binary>>},
+        Fields
+    ).
+
+caller_fields_caps_the_caller_name_test() ->
+    Name = binary:copy(<<"a">>, ?MAX_CALLER_NAME_BYTES + 32),
+    Fields = with_endpoint_env(fun() ->
+        caller_fields(#{
+            caller_id => 1234567890123456789, caller_name => Name, caller_avatar => undefined
+        })
+    end),
+    ?assertEqual(
+        ?MAX_CALLER_NAME_BYTES, byte_size(maps:get(<<"caller_name">>, Fields))
+    ).
+
+-endif.

fluxer_gateway/src/push/push_sender_delivery.erl

@@ -737,7 +737,7 @@ header_value(Name, Headers) ->
 
 capture_web_push(Payload) ->
     Endpoint = <<
-        "https://push.fluxer.app/relay/v1/apns/stable/production/",
+        "https://push.fluxer.com/relay/v1/apns/stable/production/",
         (binary:copy(<<"a">>, 64))/binary
     >>,
     {PeerPub, _PeerPriv} = crypto:generate_key(ecdh, prime256v1),

fluxer_push/src/config.rs

@@ -17,6 +17,7 @@ const DEFAULT_QUEUE_CAPACITY: usize = 10_000;
 const DEFAULT_SEND_CONCURRENCY: usize = 256;
 const DEFAULT_RELAY_MAX_CONCURRENT: usize = 1_024;
 const DEFAULT_RELAY_MAX_BODY_BYTES: usize = 2_816;
+const DEFAULT_TRUSTED_PROXY_HOPS: usize = 1;
 const DEFAULT_DEVICE_TOKEN_BUCKET_ENTRIES: usize = 1_000_000;
 const DEFAULT_DEVICE_TOKEN_BUCKET_PER_MINUTE: u32 = 60;
 const DEFAULT_DEVICE_TOKEN_BUCKET_BURST: u32 = 20;
@@ -198,6 +199,7 @@ pub struct RelayConfig {
     pub max_body_bytes: usize,
     pub trust_client_ip_header: bool,
     pub client_ip_header_name: String,
+    pub trusted_proxy_hops: usize,
     pub device_token_bucket: BucketConfig,
     pub source_bucket: Option<BucketConfig>,
     pub apns: Option<ApnsConfig>,
@@ -297,6 +299,13 @@ impl RelayConfig {
                 .get("FLUXER_CLIENT_IP_HEADER_NAME")
                 .unwrap_or(DEFAULT_CLIENT_IP_HEADER_NAME)
                 .to_ascii_lowercase(),
+            trusted_proxy_hops: parse_number(
+                "FLUXER_PUSH_RELAY_TRUSTED_PROXY_HOPS",
+                env.get("FLUXER_PUSH_RELAY_TRUSTED_PROXY_HOPS"),
+                DEFAULT_TRUSTED_PROXY_HOPS,
+                0,
+                8,
+            )?,
             device_token_bucket: bucket_config(
                 &env,
                 "FLUXER_PUSH_RELAY_TOKEN_BUCKET",

fluxer_push/src/delivery.rs

@@ -58,10 +58,17 @@ enum Audience {
 
 impl Audience {
     fn admits(self, subscription: &Subscription) -> bool {
-        let voip = subscription.platform() == Some(Platform::IosApnsVoip);
         match self {
-            Self::Standard => !voip,
-            Self::Ring => voip,
+            Self::Standard => subscription.platform() != Some(Platform::IosApnsVoip),
+            Self::Ring => Self::rings(subscription),
+        }
+    }
+
+    fn rings(subscription: &Subscription) -> bool {
+        match subscription.platform() {
+            Some(Platform::IosApnsVoip) => true,
+            Some(Platform::AndroidFcm) => subscription.is_web_push_registration(),
+            _ => false,
         }
     }
 }

fluxer_push/src/job.rs

@@ -53,6 +53,12 @@ pub struct RingJob {
     pub message_id: String,
     pub started_at_ms: i64,
     pub expires_at_ms: i64,
+    #[serde(default)]
+    pub caller_id: Option<String>,
+    #[serde(default)]
+    pub caller_name: Option<String>,
+    #[serde(default)]
+    pub caller_avatar_url: Option<String>,
 }
 
 #[derive(Debug, Error)]

fluxer_push/src/payload.rs

@@ -20,6 +20,11 @@ const SHRUNK_BODY_MAX_BYTES: usize = 40;
 const MINIMAL_TITLE_MAX_BYTES: usize = 120;
 const MEDIA_KEYS: [&str; 2] = ["image_url", "image"];
 const ICON_KEYS: [&str; 3] = ["icon", "badge", "author_avatar_url"];
+const RING_CALLER_ID_KEY: &str = "caller_id";
+const RING_CALLER_NAME_KEY: &str = "caller_name";
+const RING_CALLER_AVATAR_KEY: &str = "caller_avatar_url";
+const RING_AVATAR_KEYS: [&str; 1] = [RING_CALLER_AVATAR_KEY];
+const RING_IDENTITY_KEYS: [&str; 2] = [RING_CALLER_ID_KEY, RING_CALLER_NAME_KEY];
 const MINIMAL_DATA_KEYS: [&str; 7] = [
     "channel_id",
     "message_id",
@@ -106,13 +111,15 @@ pub fn web_push_clear(job: &ClearJob, badge_count: u32) -> Value {
 }
 
 pub fn web_push_call_ring(job: &RingJob) -> Value {
-    let data = json!({
+    let mut data = json!({
         "type": RING_TYPE,
         "channel_id": job.channel_id,
         "message_id": job.message_id,
         "target_user_id": job.user_id,
         "started_at_ms": job.started_at_ms,
+        "expires_at_ms": job.expires_at_ms,
     });
+    put_caller(&mut data, job);
     json!({
         "web_push": WEB_PUSH_MARKER,
         "type": RING_TYPE,
@@ -120,6 +127,22 @@ pub fn web_push_call_ring(job: &RingJob) -> Value {
     })
 }
 
+fn put_caller(data: &mut Value, job: &RingJob) {
+    let Some(object) = data.as_object_mut() else {
+        return;
+    };
+    let caller = [
+        (RING_CALLER_ID_KEY, job.caller_id.as_deref()),
+        (RING_CALLER_NAME_KEY, job.caller_name.as_deref()),
+        (RING_CALLER_AVATAR_KEY, job.caller_avatar_url.as_deref()),
+    ];
+    for (key, value) in caller {
+        if let Some(value) = value.filter(|value| !value.is_empty()) {
+            object.insert(key.to_owned(), value.into());
+        }
+    }
+}
+
 pub fn fcm_message(device_token: &str, envelope: &Value) -> Value {
     if is_clear(envelope) {
         return fcm_clear_message(device_token, envelope);
@@ -397,6 +420,9 @@ pub fn fit(envelope: &Value, budget: usize) -> (Vec<u8>, Option<PayloadShrink>)
     if serialized.len() <= budget {
         return (serialized, None);
     }
+    if matches!(record_kind(envelope), RecordKind::Ring) {
+        return fit_ring(envelope, budget);
+    }
     let mut working = envelope.clone();
     for step in PayloadShrink::ALL {
         working = shrink(&working, step, budget);
@@ -408,6 +434,17 @@ pub fn fit(envelope: &Value, budget: usize) -> (Vec<u8>, Option<PayloadShrink>)
     (serialize(&working), Some(PayloadShrink::Minimal))
 }
 
+fn fit_ring(envelope: &Value, budget: usize) -> (Vec<u8>, Option<PayloadShrink>) {
+    let mut working = envelope.clone();
+    drop_keys(&mut working, &RING_AVATAR_KEYS);
+    let serialized = serialize(&working);
+    if serialized.len() <= budget {
+        return (serialized, Some(PayloadShrink::Icons));
+    }
+    drop_keys(&mut working, &RING_IDENTITY_KEYS);
+    (serialize(&working), Some(PayloadShrink::Minimal))
+}
+
 fn shrink(envelope: &Value, step: PayloadShrink, budget: usize) -> Value {
     let mut working = envelope.clone();
     match step {

fluxer_push/src/providers/mod.rs

@@ -49,7 +49,10 @@ impl From<VendorOutcome> for SendOutcome {
     fn from(outcome: VendorOutcome) -> Self {
         match outcome {
             VendorOutcome::Accepted => Self::Accepted,
-            VendorOutcome::Unreachable => Self::transient("transport"),
+            VendorOutcome::Unreachable(unreachable) if unreachable.is_permanent() => {
+                Self::permanent(unreachable.label())
+            }
+            VendorOutcome::Unreachable(unreachable) => Self::transient(unreachable.label()),
             VendorOutcome::Refused(refusal) => match refusal.dead_token {
                 Some(dead_token) => Self::TokenInvalid {
                     reason: dead_token.label(),

fluxer_push/src/providers/web_push.rs

@@ -6,12 +6,13 @@ use crate::providers::SendOutcome;
 use crate::resolver;
 use crate::server::AppState;
 use crate::subscription::Subscription;
-use crate::vendor::is_transient_status;
+use crate::vendor::{Unreachable, is_transient_status};
 use rand::RngExt as _;
 use reqwest::header::{AUTHORIZATION, CONTENT_ENCODING, CONTENT_TYPE};
 use serde_json::Value;
 use std::net::IpAddr;
 use std::time::Duration;
+use tracing::warn;
 use url::{Host, Url};
 
 pub const RECORD_SIZE: usize = 2816;
@@ -90,10 +91,20 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen
 
         let status = match response {
             Ok(response) => response.status().as_u16(),
-            Err(_) if attempt >= MAX_TRANSIENT_RETRIES => {
-                return SendOutcome::transient("transport");
-            }
-            Err(_) => {
+            Err(error) => {
+                let unreachable = Unreachable::of(&error);
+                warn!(
+                    error = %error,
+                    kind = unreachable.label(),
+                    endpoint = %origin_of(&sub.endpoint),
+                    "web push request did not complete"
+                );
+                if unreachable.is_permanent() {
+                    return SendOutcome::permanent(unreachable.label());
+                }
+                if attempt >= MAX_TRANSIENT_RETRIES {
+                    return SendOutcome::transient(unreachable.label());
+                }
                 tokio::time::sleep(retry_delay(attempt)).await;
                 attempt += 1;
                 continue;

fluxer_push/src/relay/client_ip.rs

@@ -11,7 +11,7 @@ fn resolve(cfg: &RelayConfig, peer: SocketAddr, headers: &HeaderMap) -> Option<I
     headers
         .get(&cfg.client_ip_header_name)
         .and_then(|value| value.to_str().ok())
-        .and_then(nearest_entry)
+        .and_then(|value| entry_from_right(value, cfg.trusted_proxy_hops))
         .map(|ip| ip.to_canonical())
 }
 
@@ -19,9 +19,37 @@ pub fn for_rate_limit(cfg: &RelayConfig, peer: SocketAddr, headers: &HeaderMap)
     resolve(cfg, peer, headers).unwrap_or_else(|| peer.ip().to_canonical())
 }
 
-fn nearest_entry(value: &str) -> Option<IpAddr> {
+fn entry_from_right(value: &str, skip: usize) -> Option<IpAddr> {
     value
         .rsplit(',')
-        .filter_map(|entry| entry.trim().trim_matches(['[', ']']).parse().ok())
-        .next()
+        .map(|entry| entry.trim().trim_matches(['[', ']']))
+        .nth(skip)
+        .and_then(|entry| entry.parse().ok())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const EDGE: &str = "203.0.113.9";
+    const INSTANCE: &str = "198.51.100.7";
+
+    #[test]
+    fn the_rightmost_entry_is_the_edge_not_the_sending_instance() {
+        let chain = format!("{INSTANCE}, {EDGE}");
+        assert_eq!(entry_from_right(&chain, 0).unwrap().to_string(), EDGE);
+        assert_eq!(entry_from_right(&chain, 1).unwrap().to_string(), INSTANCE);
+    }
+
+    #[test]
+    fn a_longer_chain_still_names_the_sending_instance() {
+        let chain = format!("1.2.3.4, 5.6.7.8, {INSTANCE}, {EDGE}");
+        assert_eq!(entry_from_right(&chain, 1).unwrap().to_string(), INSTANCE);
+    }
+
+    #[test]
+    fn a_short_chain_yields_nothing_rather_than_a_wrong_answer() {
+        assert!(entry_from_right(EDGE, 1).is_none());
+        assert!(entry_from_right("", 1).is_none());
+    }
 }

fluxer_push/src/relay/mod.rs

@@ -37,7 +37,8 @@ const TTL_HEADER: &str = "ttl";
 const URGENCY_HEADER: &str = "urgency";
 const JSON_CONTENT_TYPE: &str = "application/json";
 const DIGEST_BYTES: usize = 8;
-const APNS_DEVICE_TOKEN_LEN: usize = 64;
+const MIN_APNS_DEVICE_TOKEN_LEN: usize = 64;
+const MAX_APNS_DEVICE_TOKEN_LEN: usize = 256;
 const MAX_FCM_DEVICE_TOKEN_LEN: usize = 512;
 const MAX_TTL_SECONDS: i64 = 86_400;
 const BODY_READ_TIMEOUT: Duration = Duration::from_secs(15);
@@ -396,7 +397,7 @@ fn finish(
     })?;
     let (result, verdict) = match outcome {
         VendorOutcome::Accepted => (RelayResult::Accepted, Ok(())),
-        VendorOutcome::Unreachable => (
+        VendorOutcome::Unreachable(_) => (
             RelayResult::Failed,
             Err(Rejection::new(Reason::ProviderUnavailable)),
         ),
@@ -425,7 +426,8 @@ fn refusal_reason(refusal: &Refusal) -> Reason {
 fn device_token_is_shaped(leg: RelayLeg, device_token: &str) -> bool {
     match leg {
         RelayLeg::Apns | RelayLeg::ApnsVoip => {
-            device_token.len() == APNS_DEVICE_TOKEN_LEN
+            (MIN_APNS_DEVICE_TOKEN_LEN..=MAX_APNS_DEVICE_TOKEN_LEN).contains(&device_token.len())
+                && device_token.len().is_multiple_of(2)
                 && device_token.bytes().all(|byte| byte.is_ascii_hexdigit())
         }
         RelayLeg::Fcm => {
@@ -485,3 +487,42 @@ fn digest(value: &str) -> String {
     }
     BASE64_URL_SAFE_NO_PAD.encode(&Sha256::digest(value.as_bytes())[..DIGEST_BYTES])
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn hex(len: usize) -> String {
+        "a".repeat(len)
+    }
+
+    #[test]
+    fn apns_accepts_every_token_length_apple_hands_out() {
+        for len in [64, 128, 160, 200, 256] {
+            assert!(
+                device_token_is_shaped(RelayLeg::Apns, &hex(len)),
+                "{len} hex characters must be accepted"
+            );
+        }
+        assert!(device_token_is_shaped(RelayLeg::Apns, &"A".repeat(64)));
+        assert!(device_token_is_shaped(RelayLeg::ApnsVoip, &hex(160)));
+    }
+
+    #[test]
+    fn apns_rejects_tokens_that_are_not_even_length_hex() {
+        for token in [hex(62), hex(63), hex(161), hex(258), "z".repeat(64)] {
+            assert!(
+                !device_token_is_shaped(RelayLeg::Apns, &token),
+                "{token} must be rejected"
+            );
+        }
+    }
+
+    #[test]
+    fn payload_too_large_answers_413() {
+        assert_eq!(
+            Reason::PayloadTooLarge.status(),
+            StatusCode::PAYLOAD_TOO_LARGE
+        );
+    }
+}

fluxer_push/src/relay/reject.rs

@@ -45,10 +45,10 @@ impl Reason {
 
     pub fn status(self) -> StatusCode {
         match self {
-            Self::BadRequest
-            | Self::PayloadTooLarge
-            | Self::DeviceTokenInvalid
-            | Self::AppUnknown => StatusCode::BAD_REQUEST,
+            Self::BadRequest | Self::DeviceTokenInvalid | Self::AppUnknown => {
+                StatusCode::BAD_REQUEST
+            }
+            Self::PayloadTooLarge => StatusCode::PAYLOAD_TOO_LARGE,
             Self::DeviceTokenGone => StatusCode::GONE,
             Self::RateLimited => StatusCode::TOO_MANY_REQUESTS,
             Self::ProviderUnavailable => StatusCode::BAD_GATEWAY,

fluxer_push/src/vendor.rs

@@ -8,6 +8,7 @@ use reqwest::header::{AUTHORIZATION, CONTENT_TYPE};
 use reqwest::redirect::Policy;
 use serde_json::Value;
 use std::time::Duration;
+use tracing::warn;
 
 const HTTP_TIMEOUT: Duration = Duration::from_secs(10);
 const APNS_TIMEOUT: Duration = Duration::from_secs(5);
@@ -15,6 +16,7 @@ const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
 const APNS_TOPIC_HEADER: &str = "apns-topic";
 pub const FCM_CONTENT_TYPE: &str = "application/json; charset=UTF-8";
 const TOO_MANY_REQUESTS: u16 = 429;
+const DNS_ERROR_MARKER: &str = "dns error";
 const MAX_ERROR_BODY_BYTES: usize = 8_192;
 const HTTP_ERROR: &str = "http_error";
 const UNREGISTERED: &str = "UNREGISTERED";
@@ -58,7 +60,45 @@ pub struct ApnsRequest<'a> {
 pub enum VendorOutcome {
     Accepted,
     Refused(Refusal),
-    Unreachable,
+    Unreachable(Unreachable),
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum Unreachable {
+    Dns,
+    Transport,
+}
+
+impl Unreachable {
+    pub fn label(self) -> &'static str {
+        match self {
+            Self::Dns => "dns",
+            Self::Transport => "transport",
+        }
+    }
+
+    pub fn is_permanent(self) -> bool {
+        matches!(self, Self::Dns)
+    }
+
+    pub fn of(error: &reqwest::Error) -> Self {
+        if names_no_host(error) {
+            Self::Dns
+        } else {
+            Self::Transport
+        }
+    }
+}
+
+fn names_no_host(error: &reqwest::Error) -> bool {
+    let mut current = std::error::Error::source(error);
+    while let Some(error) = current {
+        if error.to_string().contains(DNS_ERROR_MARKER) {
+            return true;
+        }
+        current = error.source();
+    }
+    false
 }
 
 #[derive(Debug, Eq, PartialEq)]
@@ -139,8 +179,17 @@ async fn outcome(
     response: reqwest::Result<reqwest::Response>,
     refusal: fn(u16, &[u8]) -> Refusal,
 ) -> VendorOutcome {
-    let Ok(response) = response else {
-        return VendorOutcome::Unreachable;
+    let response = match response {
+        Ok(response) => response,
+        Err(error) => {
+            let unreachable = Unreachable::of(&error);
+            warn!(
+                error = %error,
+                kind = unreachable.label(),
+                "vendor request did not complete"
+            );
+            return VendorOutcome::Unreachable(unreachable);
+        }
     };
     if response.status().is_success() {
         return VendorOutcome::Accepted;
@@ -224,3 +273,34 @@ pub async fn read_error_body(response: reqwest::Response) -> Vec<u8> {
     body.truncate(MAX_ERROR_BODY_BYTES);
     body
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    async fn error_for(url: &str) -> reqwest::Error {
+        reqwest::Client::builder()
+            .no_proxy()
+            .connect_timeout(CONNECT_TIMEOUT)
+            .build()
+            .expect("the http client builds")
+            .post(url)
+            .send()
+            .await
+            .expect_err("the request cannot complete")
+    }
+
+    #[tokio::test]
+    async fn a_host_that_does_not_resolve_is_permanent() {
+        let error = error_for("https://push.invalid/relay/v1/apns/stable/production/token").await;
+        assert_eq!(Unreachable::of(&error), Unreachable::Dns);
+        assert!(Unreachable::of(&error).is_permanent());
+    }
+
+    #[tokio::test]
+    async fn a_refused_connection_stays_retryable() {
+        let error = error_for("http://127.0.0.1:1/").await;
+        assert_eq!(Unreachable::of(&error), Unreachable::Transport);
+        assert!(!Unreachable::of(&error).is_permanent());
+    }
+}