diff --git a/api/index.ts b/api/index.ts index 5105cb2..f08ce2f 100644 --- a/api/index.ts +++ b/api/index.ts @@ -34,6 +34,8 @@ import getServices from "./so/sprk/labeler/getServices.ts"; import listNotifications from "./so/sprk/notification/listNotifications.ts"; import getUnreadCount from "./so/sprk/notification/getUnreadCount.ts"; import updateSeen from "./so/sprk/notification/updateSeen.ts"; +import registerPush from "./so/sprk/notification/registerPush.ts"; +import unregisterPush from "./so/sprk/notification/unregisterPush.ts"; export default function (server: Server, ctx: AppContext) { getAccountInfos(server, ctx); @@ -70,4 +72,6 @@ export default function (server: Server, ctx: AppContext) { listNotifications(server, ctx); getUnreadCount(server, ctx); updateSeen(server, ctx); + registerPush(server, ctx); + unregisterPush(server, ctx); } diff --git a/api/so/sprk/notification/getUnreadCount.ts b/api/so/sprk/notification/getUnreadCount.ts index 57213ab..367c3f9 100644 --- a/api/so/sprk/notification/getUnreadCount.ts +++ b/api/so/sprk/notification/getUnreadCount.ts @@ -40,11 +40,11 @@ const skeleton = async ( throw new InvalidRequestError("The seenAt parameter is unsupported"); } const priority = params.priority ?? false; - + // Get the stored lastSeenNotifs timestamp const lastSeenRes = await ctx.hydrator.dataplane.notifications .getNotificationSeen(params.viewer, priority); - + const res = await ctx.hydrator.dataplane.notifications .getUnreadNotificationCount( params.viewer, diff --git a/api/so/sprk/notification/registerPush.ts b/api/so/sprk/notification/registerPush.ts new file mode 100644 index 0000000..6551ddb --- /dev/null +++ b/api/so/sprk/notification/registerPush.ts @@ -0,0 +1,18 @@ +import { AppContext } from "../../../../context.ts"; +import { Server } from "../../../../lex/index.ts"; + +export default function (server: Server, ctx: AppContext) { + server.so.sprk.notification.registerPush({ + auth: ctx.authVerifier.standard, + handler: async ({ input, auth }) => { + const viewer = auth.credentials.iss; + await ctx.dataplane.pushTokens.upsert({ + did: viewer, + token: input.body.token, + platform: input.body.platform as "ios" | "android" | "web", + appId: input.body.appId, + serviceDid: input.body.serviceDid, + }); + }, + }); +} diff --git a/api/so/sprk/notification/unregisterPush.ts b/api/so/sprk/notification/unregisterPush.ts new file mode 100644 index 0000000..351c50c --- /dev/null +++ b/api/so/sprk/notification/unregisterPush.ts @@ -0,0 +1,12 @@ +import { AppContext } from "../../../../context.ts"; +import { Server } from "../../../../lex/index.ts"; + +export default function (server: Server, ctx: AppContext) { + server.so.sprk.notification.unregisterPush({ + auth: ctx.authVerifier.standard, + handler: async ({ input, auth }) => { + const viewer = auth.credentials.iss; + await ctx.dataplane.pushTokens.delete(viewer, input.body.token); + }, + }); +} diff --git a/config.ts b/config.ts index 8866096..9da4b7d 100644 --- a/config.ts +++ b/config.ts @@ -33,6 +33,14 @@ export interface ServerConfigValues { plcUrl?: string; labelsFromIssuerDids: string[]; + + // Push notifications + pushEnabled: boolean; + fcmServiceAccount?: string; + apnsKeyId?: string; + apnsTeamId?: string; + apnsKeyPath?: string; + apnsTopic?: string; } export class ServerConfig { @@ -76,6 +84,14 @@ export class ServerConfig { const labelsFromIssuerDids = envList("SPRK_LABELS_FROM_ISSUER_DIDS") ?? []; + // Push notifications + const pushEnabled = Deno.env.get("SPRK_PUSH_ENABLED") === "true"; + const fcmServiceAccount = envStr("SPRK_FCM_SERVICE_ACCOUNT"); + const apnsKeyId = envStr("SPRK_APNS_KEY_ID"); + const apnsTeamId = envStr("SPRK_APNS_TEAM_ID"); + const apnsKeyPath = envStr("SPRK_APNS_KEY_PATH"); + const apnsTopic = envStr("SPRK_APNS_TOPIC"); + return new ServerConfig({ version, debugMode, @@ -102,6 +118,12 @@ export class ServerConfig { relayUrl, plcUrl, labelsFromIssuerDids, + pushEnabled, + fcmServiceAccount, + apnsKeyId, + apnsTeamId, + apnsKeyPath, + apnsTopic, }); } @@ -186,4 +208,24 @@ export class ServerConfig { get labelsFromIssuerDids() { return this.cfg.labelsFromIssuerDids; } + + // Push notifications + get pushEnabled() { + return this.cfg.pushEnabled; + } + get fcmServiceAccount() { + return this.cfg.fcmServiceAccount; + } + get apnsKeyId() { + return this.cfg.apnsKeyId; + } + get apnsTeamId() { + return this.cfg.apnsTeamId; + } + get apnsKeyPath() { + return this.cfg.apnsKeyPath; + } + get apnsTopic() { + return this.cfg.apnsTopic; + } } diff --git a/data-plane/db/index.ts b/data-plane/db/index.ts index c5195d7..4607dd5 100644 --- a/data-plane/db/index.ts +++ b/data-plane/db/index.ts @@ -140,6 +140,10 @@ export class Database { "Notification", models.notificationSchema, ), + PushToken: this.connection.model( + "PushToken", + models.pushTokenSchema, + ), }; this.logger.info("Started connection to MongoDB"); diff --git a/data-plane/db/models.ts b/data-plane/db/models.ts index 1f783f8..f1a7689 100644 --- a/data-plane/db/models.ts +++ b/data-plane/db/models.ts @@ -620,6 +620,28 @@ export const notificationSchema = new Schema({ .index({ did: 1, sortAt: -1 }) .index({ did: 1, reason: 1, sortAt: -1 }); +// push tokens + +export interface PushTokenDocument extends Document { + did: string; + token: string; + platform: "ios" | "android" | "web"; + appId: string; + serviceDid: string; + createdAt: string; + updatedAt: string; +} +export const pushTokenSchema = new Schema({ + did: { type: String, required: true, index: true }, + token: { type: String, required: true }, + platform: { type: String, required: true, enum: ["ios", "android", "web"] }, + appId: { type: String, required: true }, + serviceDid: { type: String, required: true }, + createdAt: { type: String, required: true }, + updatedAt: { type: String, required: true }, +}) + .index({ did: 1, token: 1, platform: 1, appId: 1 }, { unique: true }); + // Apply plugin to schemas that extend AuthoredDocument ([ profileSchema, @@ -658,4 +680,5 @@ export interface DatabaseModels { Preference: Model; CursorState: Model; Notification: Model; + PushToken: Model; } diff --git a/data-plane/index.ts b/data-plane/index.ts index 3b5b828..4696daa 100644 --- a/data-plane/index.ts +++ b/data-plane/index.ts @@ -21,6 +21,7 @@ import { Threads } from "./routes/threads.ts"; import { Preferences } from "./routes/preferences.ts"; import { Search } from "./routes/search.ts"; import { Labels } from "./routes/labels.ts"; +import { PushTokens } from "./routes/push-tokens.ts"; export { RepoSubscription } from "./subscription.ts"; @@ -55,6 +56,7 @@ export class DataPlane { public preferences: Preferences; public search: Search; public labels: Labels; + public pushTokens: PushTokens; constructor( db: Database, @@ -85,5 +87,6 @@ export class DataPlane { this.preferences = new Preferences(db); this.search = new Search(db); this.labels = new Labels(db); + this.pushTokens = new PushTokens(db); } } diff --git a/data-plane/indexing/index.ts b/data-plane/indexing/index.ts index 854fee5..d87c248 100644 --- a/data-plane/indexing/index.ts +++ b/data-plane/indexing/index.ts @@ -29,6 +29,7 @@ import * as Labeler from "./plugins/labeler.ts"; import { RecordProcessor } from "./processor.ts"; import { getLogger, Logger } from "@logtape/logtape"; import { ServerConfig } from "../../config.ts"; +import { PushService } from "../../utils/push.ts"; export class IndexingService { records: { @@ -45,14 +46,17 @@ export class IndexingService { labeler: Labeler.PluginType; }; logger: Logger; + private pushService?: PushService; constructor( public db: Database, public cfg: ServerConfig, public idResolver: IdResolver, public background: BackgroundQueue, + pushService?: PushService, ) { this.logger = getLogger(["appview", "indexer"]); + this.pushService = pushService; this.records = { post: Post.makePlugin(this.db, this.background), reply: Reply.makePlugin(this.db, this.background), @@ -66,6 +70,13 @@ export class IndexingService { audio: Audio.makePlugin(this.db, this.background), labeler: Labeler.makePlugin(this.db, this.background), }; + + // Set push service on all processors + if (pushService) { + Object.values(this.records).forEach((processor) => { + processor.setPushService(pushService); + }); + } } transact(txn: Database) { @@ -74,6 +85,7 @@ export class IndexingService { this.cfg, this.idResolver, this.background, + this.pushService, ); } diff --git a/data-plane/indexing/processor.ts b/data-plane/indexing/processor.ts index a6b9a79..fb585ab 100644 --- a/data-plane/indexing/processor.ts +++ b/data-plane/indexing/processor.ts @@ -5,6 +5,7 @@ import { lexicons } from "../../lex/lexicons.ts"; import { BackgroundQueue } from "../background.ts"; import { Database } from "../db/index.ts"; import { chunkArray } from "@atp/common"; +import { PushService } from "../../utils/push.ts"; // @NOTE re: insertions and deletions. Due to how record updates are handled, // (insertFn) should have the same effect as (insertFn -> deleteFn -> insertFn). @@ -44,6 +45,7 @@ type Notif = { export class RecordProcessor { collection: string; db: Database; + private pushService: PushService | null = null; /** * RecordProcessor for handling a single AT Protocol collection. @@ -73,6 +75,10 @@ export class RecordProcessor { this.collection = this.params.lexId; } + setPushService(pushService: PushService) { + this.pushService = pushService; + } + matchesCollection(uri: AtUri): boolean { return uri.collection === this.collection; } @@ -336,6 +342,21 @@ export class RecordProcessor { for (const fn of runOnCommit) { await fn(this.appDb); // these could be backgrounded } + + // Queue push notifications in the background + if (this.pushService?.enabled && notifs.length > 0) { + for (const notif of notifs) { + this.background.add(async () => { + await this.pushService?.sendPush(notif.did, { + recipientDid: notif.did, + reason: notif.reason, + author: notif.author, + recordUri: notif.recordUri, + reasonSubject: notif.reasonSubject, + }); + }); + } + } } // Filter notifications for thread mutes (placeholder for future implementation) diff --git a/data-plane/routes/push-tokens.ts b/data-plane/routes/push-tokens.ts new file mode 100644 index 0000000..6ae51b1 --- /dev/null +++ b/data-plane/routes/push-tokens.ts @@ -0,0 +1,76 @@ +import { Database } from "../db/index.ts"; + +export interface PushTokenInput { + did: string; + token: string; + platform: "ios" | "android" | "web"; + appId: string; + serviceDid: string; +} + +export interface PushToken { + did: string; + token: string; + platform: "ios" | "android" | "web"; + appId: string; + serviceDid: string; + createdAt: string; + updatedAt: string; +} + +export class PushTokens { + private db: Database; + + constructor(db: Database) { + this.db = db; + } + + async upsert(input: PushTokenInput): Promise { + const now = new Date().toISOString(); + + await this.db.models.PushToken.findOneAndUpdate( + { + did: input.did, + token: input.token, + platform: input.platform, + appId: input.appId, + }, + { + $set: { + serviceDid: input.serviceDid, + updatedAt: now, + }, + $setOnInsert: { + did: input.did, + token: input.token, + platform: input.platform, + appId: input.appId, + createdAt: now, + }, + }, + { upsert: true }, + ); + } + + async delete(did: string, token: string): Promise { + await this.db.models.PushToken.deleteOne({ did, token }); + } + + async getTokensForDid(did: string): Promise { + const tokens = await this.db.models.PushToken.find({ did }).lean(); + return tokens.map((t) => ({ + did: t.did, + token: t.token, + platform: t.platform, + appId: t.appId, + serviceDid: t.serviceDid, + createdAt: t.createdAt, + updatedAt: t.updatedAt, + })); + } + + async deleteInvalidTokens(tokens: string[]): Promise { + if (tokens.length === 0) return; + await this.db.models.PushToken.deleteMany({ token: { $in: tokens } }); + } +} diff --git a/data-plane/subscription.ts b/data-plane/subscription.ts index f8bcde9..12266ff 100644 --- a/data-plane/subscription.ts +++ b/data-plane/subscription.ts @@ -6,6 +6,8 @@ import { Database } from "./db/index.ts"; import { IndexingService } from "./indexing/index.ts"; import { getLogger, Logger } from "@logtape/logtape"; import { ServerConfig } from "../config.ts"; +import { PushService } from "../utils/push.ts"; +import { PushTokens } from "./routes/push-tokens.ts"; export class RepoSubscription { firehose: Firehose; @@ -13,6 +15,7 @@ export class RepoSubscription { background: BackgroundQueue; indexingSvc: IndexingService; logger: Logger; + pushService: PushService; private firehoseRunning = false; constructor( @@ -26,11 +29,24 @@ export class RepoSubscription { const { db, idResolver, startCursor, cfg } = opts; this.logger = getLogger(["appview", "subscription"]); this.background = new BackgroundQueue(db, this.logger); + + // Create push service + const pushTokens = new PushTokens(db); + this.pushService = new PushService(pushTokens, db, { + enabled: cfg.pushEnabled, + fcmServiceAccount: cfg.fcmServiceAccount, + apnsKeyId: cfg.apnsKeyId, + apnsTeamId: cfg.apnsTeamId, + apnsKeyPath: cfg.apnsKeyPath, + apnsTopic: cfg.apnsTopic, + }); + this.indexingSvc = new IndexingService( db, cfg, idResolver, this.background, + this.pushService, ); const { runner, firehose } = createFirehose({ diff --git a/lex/index.ts b/lex/index.ts index d879d6f..23c9e74 100644 --- a/lex/index.ts +++ b/lex/index.ts @@ -173,6 +173,7 @@ import type * as SoSprkVideoGetUploadLimits from "./types/so/sprk/video/getUploa import type * as SoSprkNotificationRegisterPush from "./types/so/sprk/notification/registerPush.ts"; import type * as SoSprkNotificationPutPreferences from "./types/so/sprk/notification/putPreferences.ts"; import type * as SoSprkNotificationUpdateSeen from "./types/so/sprk/notification/updateSeen.ts"; +import type * as SoSprkNotificationUnregisterPush from "./types/so/sprk/notification/unregisterPush.ts"; import type * as SoSprkNotificationListNotifications from "./types/so/sprk/notification/listNotifications.ts"; import type * as SoSprkNotificationGetUnreadCount from "./types/so/sprk/notification/getUnreadCount.ts"; import type * as SoSprkGraphGetSuggestedFollowsByActor from "./types/so/sprk/graph/getSuggestedFollowsByActor.ts"; @@ -2751,6 +2752,18 @@ export class SoSprkNotificationNS { return this._server.xrpc.method(nsid, cfg); } + unregisterPush( + cfg: MethodConfigOrHandler< + A, + SoSprkNotificationUnregisterPush.QueryParams, + SoSprkNotificationUnregisterPush.HandlerInput, + SoSprkNotificationUnregisterPush.HandlerOutput + >, + ) { + const nsid = "so.sprk.notification.unregisterPush"; // @ts-ignore - dynamically generated + return this._server.xrpc.method(nsid, cfg); + } + listNotifications( cfg: MethodConfigOrHandler< A, diff --git a/lex/lexicons.ts b/lex/lexicons.ts index 89d6992..1694d27 100644 --- a/lex/lexicons.ts +++ b/lex/lexicons.ts @@ -16375,6 +16375,49 @@ export const schemaDict = { }, }, }, + "SoSprkNotificationUnregisterPush": { + "lexicon": 1, + "id": "so.sprk.notification.unregisterPush", + "defs": { + "main": { + "type": "procedure", + "description": + "The inverse of registerPush - inform a specified service that push notifications should no longer be sent to the given token for the requesting account. Requires auth.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": [ + "serviceDid", + "token", + "platform", + "appId", + ], + "properties": { + "serviceDid": { + "type": "string", + "format": "did", + }, + "token": { + "type": "string", + }, + "platform": { + "type": "string", + "knownValues": [ + "ios", + "android", + "web", + ], + }, + "appId": { + "type": "string", + }, + }, + }, + }, + }, + }, + }, "SoSprkNotificationListNotifications": { "lexicon": 1, "id": "so.sprk.notification.listNotifications", @@ -16477,8 +16520,8 @@ export const schemaDict = { "follow", "mention", "reply", - "quote", - "starterpack-joined", + "like-via-repost", + "repost-via-repost", ], }, "reasonSubject": { @@ -26785,6 +26828,7 @@ export const ids = { SoSprkNotificationRegisterPush: "so.sprk.notification.registerPush", SoSprkNotificationPutPreferences: "so.sprk.notification.putPreferences", SoSprkNotificationUpdateSeen: "so.sprk.notification.updateSeen", + SoSprkNotificationUnregisterPush: "so.sprk.notification.unregisterPush", SoSprkNotificationListNotifications: "so.sprk.notification.listNotifications", SoSprkNotificationGetUnreadCount: "so.sprk.notification.getUnreadCount", SoSprkGraphGetSuggestedFollowsByActor: diff --git a/lex/types/so/sprk/notification/listNotifications.ts b/lex/types/so/sprk/notification/listNotifications.ts index 922d04f..2c6af6e 100644 --- a/lex/types/so/sprk/notification/listNotifications.ts +++ b/lex/types/so/sprk/notification/listNotifications.ts @@ -53,8 +53,8 @@ export interface Notification { | "follow" | "mention" | "reply" - | "quote" - | "starterpack-joined" + | "like-via-repost" + | "repost-via-repost" | (string & globalThis.Record); reasonSubject?: string; record: { [_ in string]: unknown }; diff --git a/lex/types/so/sprk/notification/unregisterPush.ts b/lex/types/so/sprk/notification/unregisterPush.ts new file mode 100644 index 0000000..5687ba5 --- /dev/null +++ b/lex/types/so/sprk/notification/unregisterPush.ts @@ -0,0 +1,27 @@ +/** + * GENERATED CODE - DO NOT MODIFY + */ +export type QueryParams = globalThis.Record; + +export interface InputSchema { + serviceDid: string; + token: string; + platform: + | "ios" + | "android" + | "web" + | (string & globalThis.Record); + appId: string; +} + +export interface HandlerInput { + encoding: "application/json"; + body: InputSchema; +} + +export interface HandlerError { + status: number; + message?: string; +} + +export type HandlerOutput = HandlerError | void; diff --git a/lexicons/so/sprk/notification/listNotifications.json b/lexicons/so/sprk/notification/listNotifications.json index 8afd318..97e143d 100644 --- a/lexicons/so/sprk/notification/listNotifications.json +++ b/lexicons/so/sprk/notification/listNotifications.json @@ -68,8 +68,8 @@ "follow", "mention", "reply", - "quote", - "starterpack-joined" + "like-via-repost", + "repost-via-repost" ] }, "reasonSubject": { "type": "string", "format": "at-uri" }, diff --git a/lexicons/so/sprk/notification/unregisterPush.json b/lexicons/so/sprk/notification/unregisterPush.json new file mode 100644 index 0000000..31c2726 --- /dev/null +++ b/lexicons/so/sprk/notification/unregisterPush.json @@ -0,0 +1,26 @@ +{ + "lexicon": 1, + "id": "so.sprk.notification.unregisterPush", + "defs": { + "main": { + "type": "procedure", + "description": "The inverse of registerPush - inform a specified service that push notifications should no longer be sent to the given token for the requesting account. Requires auth.", + "input": { + "encoding": "application/json", + "schema": { + "type": "object", + "required": ["serviceDid", "token", "platform", "appId"], + "properties": { + "serviceDid": { "type": "string", "format": "did" }, + "token": { "type": "string" }, + "platform": { + "type": "string", + "knownValues": ["ios", "android", "web"] + }, + "appId": { "type": "string" } + } + } + } + } + } +} diff --git a/tests/util.ts b/tests/util.ts index 8437020..8c2a27c 100644 --- a/tests/util.ts +++ b/tests/util.ts @@ -79,6 +79,7 @@ const DEFAULT_TEST_CONFIG: ServerConfigValues = { maxThreadParents: 10, labelsFromIssuerDids: [], notificationsDelayMs: 1000, + pushEnabled: false, }; // ============================================================================ @@ -202,6 +203,10 @@ export async function createTestDatabase( "Notification", models.notificationSchema, ), + PushToken: connection.model( + "PushToken", + models.pushTokenSchema, + ), }; // Seed data diff --git a/utils/push.ts b/utils/push.ts new file mode 100644 index 0000000..0f9d2f5 --- /dev/null +++ b/utils/push.ts @@ -0,0 +1,496 @@ +import { getLogger, Logger } from "@logtape/logtape"; +import { jsonStringToLex } from "@atp/lexicon"; +import { PushToken, PushTokens } from "../data-plane/routes/push-tokens.ts"; +import { Database } from "../data-plane/db/index.ts"; + +export interface PushPayload { + recipientDid: string; + reason: string; + author: string; + recordUri: string; + reasonSubject?: string; +} + +export interface PushConfig { + enabled: boolean; + fcmServiceAccount?: string; // JSON string of Firebase service account + apnsKeyId?: string; + apnsTeamId?: string; + apnsKeyPath?: string; + apnsTopic?: string; // Bundle ID for iOS app +} + +interface FcmServiceAccount { + project_id: string; + private_key: string; + client_email: string; +} + +export class PushService { + private logger: Logger; + private pushTokens: PushTokens; + private db: Database; + private config: PushConfig; + private fcmAccessToken: string | null = null; + private fcmTokenExpiry: number = 0; + private fcmServiceAccount: FcmServiceAccount | null = null; + private apnsPrivateKey: CryptoKey | null = null; + + constructor(pushTokens: PushTokens, db: Database, config: PushConfig) { + this.logger = getLogger(["appview", "push"]); + this.pushTokens = pushTokens; + this.db = db; + this.config = config; + + if (config.fcmServiceAccount) { + try { + this.fcmServiceAccount = JSON.parse(config.fcmServiceAccount); + } catch { + this.logger.error("Failed to parse FCM service account JSON"); + } + } + } + + get enabled(): boolean { + return this.config.enabled; + } + + async sendPush(did: string, payload: PushPayload): Promise { + if (!this.config.enabled) { + return; + } + + const tokens = await this.pushTokens.getTokensForDid(did); + if (tokens.length === 0) { + return; + } + + const invalidTokens: string[] = []; + + for (const token of tokens) { + try { + if (token.platform === "ios") { + const success = await this.sendApns(token, payload); + if (!success) { + invalidTokens.push(token.token); + } + } else if (token.platform === "android") { + const success = await this.sendFcm(token, payload); + if (!success) { + invalidTokens.push(token.token); + } + } + } catch (err) { + this.logger.error("Failed to send push notification", { + err, + platform: token.platform, + did, + }); + } + } + + // Clean up invalid tokens + if (invalidTokens.length > 0) { + await this.pushTokens.deleteInvalidTokens(invalidTokens); + this.logger.info("Removed invalid push tokens", { + count: invalidTokens.length, + }); + } + } + + private async sendFcm( + token: PushToken, + payload: PushPayload, + ): Promise { + if (!this.fcmServiceAccount) { + this.logger.warn("FCM service account not configured"); + return true; // Don't mark as invalid if not configured + } + + const accessToken = await this.getFcmAccessToken(); + if (!accessToken) { + return true; // Don't mark as invalid if we can't get a token + } + + const notification = await this.buildNotificationContent(payload); + const message = { + message: { + token: token.token, + notification: { + title: notification.title, + body: notification.body, + }, + data: { + reason: payload.reason, + author: payload.author, + recordUri: payload.recordUri, + ...(payload.reasonSubject && + { reasonSubject: payload.reasonSubject }), + }, + android: { + priority: "high" as const, + }, + }, + }; + + const projectId = this.fcmServiceAccount.project_id; + const url = + `https://fcm.googleapis.com/v1/projects/${projectId}/messages:send`; + + try { + const response = await fetch(url, { + method: "POST", + headers: { + "Authorization": `Bearer ${accessToken}`, + "Content-Type": "application/json", + }, + body: JSON.stringify(message), + }); + + if (!response.ok) { + const error = await response.json(); + // Check for unregistered token error + if ( + error.error?.details?.some( + (d: { errorCode?: string }) => + d.errorCode === "UNREGISTERED" || + d.errorCode === "INVALID_ARGUMENT", + ) + ) { + return false; // Mark as invalid + } + this.logger.error("FCM request failed", { + error, + status: response.status, + }); + } + + return true; + } catch (err) { + this.logger.error("FCM request error", { err }); + return true; // Don't mark as invalid on network errors + } + } + + private async sendApns( + token: PushToken, + payload: PushPayload, + ): Promise { + if ( + !this.config.apnsKeyId || !this.config.apnsTeamId || + !this.config.apnsKeyPath + ) { + this.logger.warn("APNs not fully configured"); + return true; // Don't mark as invalid if not configured + } + + const jwt = await this.getApnsJwt(); + if (!jwt) { + return true; // Don't mark as invalid if we can't get a JWT + } + + const notification = await this.buildNotificationContent(payload); + const apnsPayload = { + aps: { + alert: { + title: notification.title, + body: notification.body, + }, + sound: "default", + badge: 1, + }, + reason: payload.reason, + author: payload.author, + recordUri: payload.recordUri, + ...(payload.reasonSubject && { reasonSubject: payload.reasonSubject }), + }; + + const topic = this.config.apnsTopic || token.appId; + const url = `https://api.push.apple.com/3/device/${token.token}`; + + try { + const response = await fetch(url, { + method: "POST", + headers: { + "authorization": `bearer ${jwt}`, + "apns-topic": topic, + "apns-push-type": "alert", + "apns-priority": "10", + }, + body: JSON.stringify(apnsPayload), + }); + + if (!response.ok) { + const status = response.status; + // 400 = Bad device token, 410 = Token is no longer active + if (status === 400 || status === 410) { + return false; // Mark as invalid + } + this.logger.error("APNs request failed", { status }); + } + + return true; + } catch (err) { + this.logger.error("APNs request error", { err }); + return true; // Don't mark as invalid on network errors + } + } + + private async buildNotificationContent( + payload: PushPayload, + ): Promise<{ title: string; body: string }> { + // Get author handle + const author = await this.db.models.Actor.findOne({ + did: payload.author, + }).lean(); + const handle = author?.handle ? `${author.handle}` : "Someone"; + + // Handle follow notifications specially + if (payload.reason === "follow") { + // Check if recipient follows the author back (making this a "followed you back") + const recipientFollowsAuthor = await this.db.models.Follow.findOne({ + authorDid: payload.recipientDid, + subject: payload.author, + }).lean(); + + const body = recipientFollowsAuthor + ? `${handle} followed you back` + : `${handle} followed you`; + + return { + title: "New Follower", + body, + }; + } + + // Build title based on reason + const reasonMap: Record = { + like: "liked your post", + repost: "reposted your post", + mention: "mentioned you", + reply: "replied to your post", + "like-via-repost": "liked your repost", + "repost-via-repost": "reposted your repost", + }; + + const action = reasonMap[payload.reason] || "interacted with your content"; + const title = `${handle} ${action}`; + + // Build body based on reason type + let body = ""; + + if ( + payload.reason === "like" || payload.reason === "repost" || + payload.reason === "like-via-repost" || + payload.reason === "repost-via-repost" + ) { + // For likes/reposts, show the reasonSubject (the post that was liked/reposted) + if (payload.reasonSubject) { + body = await this.getRecordText(payload.reasonSubject); + } + } else if (payload.reason === "reply" || payload.reason === "mention") { + // For replies/mentions, show the record text (the reply or post with mention) + body = await this.getRecordText(payload.recordUri); + } + + return { title, body }; + } + + private async getRecordText(uri: string): Promise { + try { + const record = await this.db.models.Record.findOne({ uri }).lean(); + if (!record?.json) return ""; + + const parsed = jsonStringToLex(record.json) as { + text?: string; + caption?: { text?: string }; + }; + + // Try to get text from different record formats + const text = parsed.text || parsed.caption?.text || ""; + + // Truncate to reasonable length for push notification + if (text.length > 100) { + return text.substring(0, 97) + "..."; + } + return text; + } catch { + return ""; + } + } + + private async getFcmAccessToken(): Promise { + if (!this.fcmServiceAccount) { + return null; + } + + // Return cached token if still valid + if (this.fcmAccessToken && Date.now() < this.fcmTokenExpiry - 60000) { + return this.fcmAccessToken; + } + + try { + const now = Math.floor(Date.now() / 1000); + const exp = now + 3600; // 1 hour + + const header = { + alg: "RS256", + typ: "JWT", + }; + + const claim = { + iss: this.fcmServiceAccount.client_email, + scope: "https://www.googleapis.com/auth/firebase.messaging", + aud: "https://oauth2.googleapis.com/token", + iat: now, + exp: exp, + }; + + // Create JWT + const encoder = new TextEncoder(); + const headerB64 = this.base64UrlEncode( + encoder.encode(JSON.stringify(header)), + ); + const claimB64 = this.base64UrlEncode( + encoder.encode(JSON.stringify(claim)), + ); + const unsignedJwt = `${headerB64}.${claimB64}`; + + // Import private key and sign + const privateKey = await this.importPrivateKey( + this.fcmServiceAccount.private_key, + ); + const signature = await crypto.subtle.sign( + { name: "RSASSA-PKCS1-v1_5" }, + privateKey, + encoder.encode(unsignedJwt), + ); + + const signatureB64 = this.base64UrlEncode(new Uint8Array(signature)); + const jwt = `${unsignedJwt}.${signatureB64}`; + + // Exchange JWT for access token + const response = await fetch("https://oauth2.googleapis.com/token", { + method: "POST", + headers: { + "Content-Type": "application/x-www-form-urlencoded", + }, + body: new URLSearchParams({ + grant_type: "urn:ietf:params:oauth:grant-type:jwt-bearer", + assertion: jwt, + }), + }); + + if (!response.ok) { + this.logger.error("Failed to get FCM access token", { + status: response.status, + }); + return null; + } + + const data = await response.json(); + this.fcmAccessToken = data.access_token; + this.fcmTokenExpiry = Date.now() + (data.expires_in * 1000); + + return this.fcmAccessToken; + } catch (err) { + this.logger.error("Error getting FCM access token", { err }); + return null; + } + } + + private async getApnsJwt(): Promise { + if ( + !this.config.apnsKeyId || !this.config.apnsTeamId || + !this.config.apnsKeyPath + ) { + return null; + } + + try { + // Load APNs private key if not already loaded + if (!this.apnsPrivateKey) { + const keyData = await Deno.readTextFile(this.config.apnsKeyPath); + this.apnsPrivateKey = await this.importApnsKey(keyData); + } + + const now = Math.floor(Date.now() / 1000); + const header = { + alg: "ES256", + kid: this.config.apnsKeyId, + }; + + const claim = { + iss: this.config.apnsTeamId, + iat: now, + }; + + const encoder = new TextEncoder(); + const headerB64 = this.base64UrlEncode( + encoder.encode(JSON.stringify(header)), + ); + const claimB64 = this.base64UrlEncode( + encoder.encode(JSON.stringify(claim)), + ); + const unsignedJwt = `${headerB64}.${claimB64}`; + + const signature = await crypto.subtle.sign( + { name: "ECDSA", hash: "SHA-256" }, + this.apnsPrivateKey, + encoder.encode(unsignedJwt), + ); + + // Convert DER signature to raw format for JWT + const signatureB64 = this.base64UrlEncode(new Uint8Array(signature)); + return `${unsignedJwt}.${signatureB64}`; + } catch (err) { + this.logger.error("Error creating APNs JWT", { err }); + return null; + } + } + + private async importPrivateKey(pem: string): Promise { + const pemContents = pem + .replace("-----BEGIN PRIVATE KEY-----", "") + .replace("-----END PRIVATE KEY-----", "") + .replace(/\n/g, ""); + + const binaryDer = Uint8Array.from( + atob(pemContents), + (c) => c.charCodeAt(0), + ); + + return await crypto.subtle.importKey( + "pkcs8", + binaryDer, + { name: "RSASSA-PKCS1-v1_5", hash: "SHA-256" }, + false, + ["sign"], + ); + } + + private async importApnsKey(pem: string): Promise { + const pemContents = pem + .replace("-----BEGIN PRIVATE KEY-----", "") + .replace("-----END PRIVATE KEY-----", "") + .replace(/\n/g, ""); + + const binaryDer = Uint8Array.from( + atob(pemContents), + (c) => c.charCodeAt(0), + ); + + return await crypto.subtle.importKey( + "pkcs8", + binaryDer, + { name: "ECDSA", namedCurve: "P-256" }, + false, + ["sign"], + ); + } + + private base64UrlEncode(data: Uint8Array): string { + const base64 = btoa(String.fromCharCode(...data)); + return base64.replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, ""); + } +}