[READ ONLY MIRROR] Spark Social AppView Server
github.com/sprksocial/server
atproto
deno
hono
lexicon
feat: notifications
+822
-7
+6
api/index.ts
+6
api/index.ts
···
31
31
import getFeedGenerator from "./so/sprk/feed/getFeedGenerator.ts";
32
32
import getFeedGenerators from "./so/sprk/feed/getFeedGenerators.ts";
33
33
import getServices from "./so/sprk/labeler/getServices.ts";
34
+
import listNotifications from "./so/sprk/notification/listNotifications.ts";
35
+
import getUnreadCount from "./so/sprk/notification/getUnreadCount.ts";
36
+
import updateSeen from "./so/sprk/notification/updateSeen.ts";
34
37
35
38
export default function (server: Server, ctx: AppContext) {
36
39
getAccountInfos(server, ctx);
···
64
67
getFeedGenerator(server, ctx);
65
68
getFeedGenerators(server, ctx);
66
69
getServices(server, ctx);
70
+
listNotifications(server, ctx);
71
+
getUnreadCount(server, ctx);
72
+
updateSeen(server, ctx);
67
73
}
+78
api/so/sprk/notification/getUnreadCount.ts
+78
api/so/sprk/notification/getUnreadCount.ts
···
1
+
import { InvalidRequestError } from "@atp/xrpc-server";
2
+
import { AppContext } from "../../../../context.ts";
3
+
import { Hydrator } from "../../../../hydration/index.ts";
4
+
import { Server } from "../../../../lex/index.ts";
5
+
import { QueryParams } from "../../../../lex/types/so/sprk/notification/getUnreadCount.ts";
6
+
import {
7
+
createPipeline,
8
+
HydrationFnInput,
9
+
noRules,
10
+
PresentationFnInput,
11
+
SkeletonFnInput,
12
+
} from "../../../../pipeline.ts";
13
+
import { Views } from "../../../../views/index.ts";
14
+
15
+
export default function (server: Server, ctx: AppContext) {
16
+
const getUnreadCount = createPipeline(
17
+
skeleton,
18
+
hydration,
19
+
noRules,
20
+
presentation,
21
+
);
22
+
server.app.bsky.notification.getUnreadCount({
23
+
auth: ctx.authVerifier.standard,
24
+
handler: async ({ auth, params }) => {
25
+
const viewer = auth.credentials.iss;
26
+
const result = await getUnreadCount({ ...params, viewer }, ctx);
27
+
return {
28
+
encoding: "application/json",
29
+
body: result,
30
+
};
31
+
},
32
+
});
33
+
}
34
+
35
+
const skeleton = async (
36
+
input: SkeletonFnInput<Context, Params>,
37
+
): Promise<SkeletonState> => {
38
+
const { params, ctx } = input;
39
+
if (params.seenAt) {
40
+
throw new InvalidRequestError("The seenAt parameter is unsupported");
41
+
}
42
+
const priority = params.priority ?? false;
43
+
const res = await ctx.hydrator.dataplane.notifications
44
+
.getUnreadNotificationCount(
45
+
params.viewer,
46
+
undefined,
47
+
priority,
48
+
);
49
+
return {
50
+
count: res.count,
51
+
};
52
+
};
53
+
54
+
const hydration = (
55
+
_input: HydrationFnInput<Context, Params, SkeletonState>,
56
+
) => {
57
+
return Promise.resolve({});
58
+
};
59
+
60
+
const presentation = (
61
+
input: PresentationFnInput<Context, Params, SkeletonState>,
62
+
) => {
63
+
const { skeleton } = input;
64
+
return { count: skeleton.count };
65
+
};
66
+
67
+
type Context = {
68
+
hydrator: Hydrator;
69
+
views: Views;
70
+
};
71
+
72
+
type Params = QueryParams & {
73
+
viewer: string;
74
+
};
75
+
76
+
type SkeletonState = {
77
+
count: number;
78
+
};
+227
api/so/sprk/notification/listNotifications.ts
+227
api/so/sprk/notification/listNotifications.ts
···
1
+
import { mapDefined } from "@atp/common";
2
+
import { InvalidRequestError } from "@atp/xrpc-server";
3
+
import { ServerConfig } from "../../../../config.ts";
4
+
import { AppContext } from "../../../../context.ts";
5
+
import { Notification } from "../../../../data-plane/routes/notifs.ts";
6
+
import { HydrateCtx, Hydrator } from "../../../../hydration/index.ts";
7
+
import { Server } from "../../../../lex/index.ts";
8
+
import { QueryParams } from "../../../../lex/types/so/sprk/notification/listNotifications.ts";
9
+
import {
10
+
createPipeline,
11
+
HydrationFnInput,
12
+
PresentationFnInput,
13
+
RulesFnInput,
14
+
SkeletonFnInput,
15
+
} from "../../../../pipeline.ts";
16
+
import { uriToDid as didFromUri } from "../../../../utils/uris.ts";
17
+
import { Views } from "../../../../views/index.ts";
18
+
import { resHeaders } from "../../../util.ts";
19
+
20
+
export default function (server: Server, ctx: AppContext) {
21
+
const listNotifications = createPipeline(
22
+
skeleton,
23
+
hydration,
24
+
noBlockOrMutesOrNeedsFiltering,
25
+
presentation,
26
+
);
27
+
server.so.sprk.notification.listNotifications({
28
+
auth: ctx.authVerifier.standard,
29
+
handler: async ({ params, auth, req }) => {
30
+
const viewer = auth.credentials.iss;
31
+
const labelers = ctx.reqLabelers(req);
32
+
const hydrateCtx = await ctx.hydrator.createContext({ labelers, viewer });
33
+
const result = await listNotifications(
34
+
{ ...params, hydrateCtx: hydrateCtx.copy({ viewer }) },
35
+
ctx,
36
+
);
37
+
return {
38
+
encoding: "application/json",
39
+
body: result,
40
+
headers: resHeaders({ labelers: hydrateCtx.labelers }),
41
+
};
42
+
},
43
+
});
44
+
}
45
+
46
+
const paginateNotifications = async (opts: {
47
+
ctx: Context;
48
+
priority: boolean;
49
+
reasons?: string[];
50
+
cursor?: string;
51
+
limit: number;
52
+
viewer: string;
53
+
}) => {
54
+
const { ctx, priority, reasons, limit, viewer } = opts;
55
+
56
+
// if not filtering, then just pass through the response from dataplane
57
+
if (!reasons) {
58
+
const res = await ctx.hydrator.dataplane.notifications.getNotifications(
59
+
viewer,
60
+
limit,
61
+
opts.cursor,
62
+
priority,
63
+
);
64
+
return {
65
+
notifications: res.notifications,
66
+
cursor: res.cursor,
67
+
};
68
+
}
69
+
70
+
let nextCursor: string | undefined = opts.cursor;
71
+
let toReturn: Notification[] = [];
72
+
const maxAttempts = 10;
73
+
const attemptSize = Math.ceil(limit / 2);
74
+
for (let i = 0; i < maxAttempts; i++) {
75
+
const res = await ctx.hydrator.dataplane.notifications.getNotifications(
76
+
viewer,
77
+
limit,
78
+
nextCursor,
79
+
priority,
80
+
);
81
+
const filtered = res.notifications.filter((notif) =>
82
+
reasons.includes(notif.reason)
83
+
);
84
+
toReturn = [...toReturn, ...filtered];
85
+
nextCursor = res.cursor ?? undefined;
86
+
if (toReturn.length >= attemptSize || !nextCursor) {
87
+
break;
88
+
}
89
+
}
90
+
return {
91
+
notifications: toReturn,
92
+
cursor: nextCursor,
93
+
};
94
+
};
95
+
96
+
/**
97
+
* Applies a configurable delay to the datetime string of a cursor,
98
+
* effectively allowing for a delay on listing the notifications.
99
+
* This is useful to allow time for services to process notifications
100
+
* before they are listed to the user.
101
+
*/
102
+
export const delayCursor = (
103
+
cursorStr: string | undefined,
104
+
delayMs: number,
105
+
): string => {
106
+
const nowMinusDelay = Date.now() - delayMs;
107
+
if (cursorStr === undefined) return new Date(nowMinusDelay).toISOString();
108
+
const cursor = new Date(cursorStr).getTime();
109
+
if (isNaN(cursor)) return cursorStr;
110
+
return new Date(Math.min(cursor, nowMinusDelay)).toISOString();
111
+
};
112
+
113
+
const skeleton = async (
114
+
input: SkeletonFnInput<Context, Params>,
115
+
): Promise<SkeletonState> => {
116
+
const { params, ctx } = input;
117
+
if (params.seenAt) {
118
+
throw new InvalidRequestError("The seenAt parameter is unsupported");
119
+
}
120
+
121
+
const originalCursor = params.cursor;
122
+
const delayedCursor = delayCursor(
123
+
originalCursor,
124
+
ctx.cfg.notificationsDelayMs,
125
+
);
126
+
const viewer = params.hydrateCtx.viewer;
127
+
const priority = params.priority ?? false;
128
+
const [res, lastSeenRes] = await Promise.all([
129
+
paginateNotifications({
130
+
ctx,
131
+
priority,
132
+
reasons: params.reasons,
133
+
cursor: delayedCursor,
134
+
limit: params.limit,
135
+
viewer,
136
+
}),
137
+
ctx.hydrator.dataplane.notifications.getNotificationSeen(
138
+
viewer,
139
+
priority,
140
+
),
141
+
]);
142
+
// @NOTE for the first page of results if there's no last-seen time, consider top notification unread
143
+
// rather than all notifications. bit of a hack to be more graceful when seen times are out of sync.
144
+
let lastSeenAt = lastSeenRes.timestamp;
145
+
if (!lastSeenAt && !originalCursor) {
146
+
lastSeenAt = res.notifications.at(0)?.sortAt;
147
+
}
148
+
return {
149
+
notifs: res.notifications,
150
+
cursor: res.cursor || undefined,
151
+
priority,
152
+
lastSeenNotifs: lastSeenAt,
153
+
};
154
+
};
155
+
156
+
const hydration = async (
157
+
input: HydrationFnInput<Context, Params, SkeletonState>,
158
+
) => {
159
+
const { skeleton, params, ctx } = input;
160
+
return await ctx.hydrator.hydrateNotifications(
161
+
skeleton.notifs,
162
+
params.hydrateCtx,
163
+
);
164
+
};
165
+
166
+
const noBlockOrMutesOrNeedsFiltering = (
167
+
input: RulesFnInput<Context, Params, SkeletonState>,
168
+
) => {
169
+
const { skeleton, hydration, ctx } = input;
170
+
skeleton.notifs = skeleton.notifs.filter((item) => {
171
+
const did = didFromUri(item.uri);
172
+
if (
173
+
ctx.views.viewerBlockExists(did, hydration) ||
174
+
ctx.views.viewerMuteExists(did, hydration)
175
+
) {
176
+
return false;
177
+
}
178
+
// Filter out notifications from users that need review unless moots
179
+
if (
180
+
item.reason === "reply" ||
181
+
item.reason === "quote" ||
182
+
item.reason === "mention" ||
183
+
item.reason === "like" ||
184
+
item.reason === "follow"
185
+
) {
186
+
if (!ctx.views.viewerSeesNeedsReview({ did, uri: item.uri }, hydration)) {
187
+
return false;
188
+
}
189
+
}
190
+
return true;
191
+
});
192
+
return skeleton;
193
+
};
194
+
195
+
const presentation = (
196
+
input: PresentationFnInput<Context, Params, SkeletonState>,
197
+
) => {
198
+
const { skeleton, hydration, ctx } = input;
199
+
const { notifs, lastSeenNotifs, cursor } = skeleton;
200
+
const notifications = mapDefined(
201
+
notifs,
202
+
(notif) => ctx.views.notification(notif, lastSeenNotifs, hydration),
203
+
);
204
+
return {
205
+
notifications,
206
+
cursor,
207
+
priority: skeleton.priority,
208
+
seenAt: skeleton.lastSeenNotifs,
209
+
};
210
+
};
211
+
212
+
type Context = {
213
+
hydrator: Hydrator;
214
+
views: Views;
215
+
cfg: ServerConfig;
216
+
};
217
+
218
+
type Params = QueryParams & {
219
+
hydrateCtx: HydrateCtx & { viewer: string };
220
+
};
221
+
222
+
type SkeletonState = {
223
+
notifs: Notification[];
224
+
priority: boolean;
225
+
lastSeenNotifs?: string;
226
+
cursor?: string;
227
+
};
+26
api/so/sprk/notification/updateSeen.ts
+26
api/so/sprk/notification/updateSeen.ts
···
1
+
import { AppContext } from "../../../../context.ts";
2
+
import { Server } from "../../../../lex/index.ts";
3
+
4
+
export default function (server: Server, ctx: AppContext) {
5
+
server.so.sprk.notification.updateSeen({
6
+
auth: ctx.authVerifier.standard,
7
+
handler: async ({ input, auth }) => {
8
+
const viewer = auth.credentials.iss;
9
+
const seenAt = input.body.seenAt;
10
+
// For now we keep separate seen times behind the scenes for priority,
11
+
// but treat them as a single seen time.
12
+
await Promise.all([
13
+
ctx.hydrator.dataplane.notifications.updateNotificationSeen(
14
+
viewer,
15
+
seenAt,
16
+
false,
17
+
),
18
+
ctx.hydrator.dataplane.notifications.updateNotificationSeen(
19
+
viewer,
20
+
seenAt,
21
+
true,
22
+
),
23
+
]);
24
+
},
25
+
});
26
+
}
+6
config.ts
+6
config.ts
···
19
19
bigThreadDepth?: number;
20
20
maxThreadDepth?: number;
21
21
maxThreadParents: number;
22
+
notificationsDelayMs: number;
22
23
23
24
videoCdn?: string;
24
25
mediaCdn?: string;
···
58
59
const bigThreadDepth = envInt("SPRK_BIG_THREAD_DEPTH") ?? 10;
59
60
const maxThreadDepth = envInt("SPRK_MAX_THREAD_DEPTH") ?? 10;
60
61
const maxThreadParents = envInt("SPRK_MAX_THREAD_PARENTS") ?? 10;
62
+
const notificationsDelayMs = envInt("SPRK_NOTIFICATIONS_DELAY_MS") ?? 10;
61
63
62
64
const videoCdn = envStr("SPRK_VIDEO_CDN") ?? "https://video.sprk.so";
63
65
const mediaCdn = envStr("SPRK_MEDIA_CDN") ?? "https://media.sprk.so";
···
89
91
bigThreadDepth,
90
92
maxThreadDepth,
91
93
maxThreadParents,
94
+
notificationsDelayMs,
92
95
videoCdn,
93
96
mediaCdn,
94
97
thumbCdn,
···
145
148
}
146
149
get maxThreadDepth() {
147
150
return this.cfg.maxThreadDepth;
151
+
}
152
+
get notificationsDelayMs() {
153
+
return this.cfg.notificationsDelayMs;
148
154
}
149
155
150
156
// CDNs
+4
data-plane/db/index.ts
+4
data-plane/db/index.ts
+24
data-plane/db/models.ts
+24
data-plane/db/models.ts
···
595
595
updatedAt: { type: Date, default: Date.now },
596
596
});
597
597
598
+
// notifications
599
+
600
+
export interface NotificationDocument extends Document {
601
+
did: string;
602
+
recordUri: string;
603
+
recordCid: string;
604
+
author: string;
605
+
reason: string;
606
+
reasonSubject: string | null;
607
+
sortAt: string;
608
+
}
609
+
export const notificationSchema = new Schema<NotificationDocument>({
610
+
did: { type: String, required: true, index: true },
611
+
recordUri: { type: String, required: true, index: true },
612
+
recordCid: { type: String, required: true },
613
+
author: { type: String, required: true, index: true },
614
+
reason: { type: String, required: true },
615
+
reasonSubject: { type: String, required: false, default: null },
616
+
sortAt: { type: String, required: true, index: true },
617
+
})
618
+
.index({ did: 1, sortAt: -1 })
619
+
.index({ did: 1, reason: 1, sortAt: -1 });
620
+
598
621
// Apply plugin to schemas that extend AuthoredDocument
599
622
([
600
623
profileSchema,
···
632
655
ActorSync: Model<ActorSyncDocument>;
633
656
Preference: Model<PreferenceDocument>;
634
657
CursorState: Model<CursorStateDocument>;
658
+
Notification: Model<NotificationDocument>;
635
659
}
+3
data-plane/index.ts
+3
data-plane/index.ts
···
9
9
import { Moderation } from "./routes/moderation.ts";
10
10
import { Actors } from "./routes/actors.ts";
11
11
import { Identity } from "./routes/identity.ts";
12
+
import { Notifications } from "./routes/notifs.ts";
12
13
import { Records } from "./routes/records.ts";
13
14
import { Relationships } from "./routes/relationships.ts";
14
15
import { Interactions } from "./routes/interactions.ts";
···
42
43
public moderation: Moderation;
43
44
public actors: Actors;
44
45
public identity: Identity;
46
+
public notifications: Notifications;
45
47
public records: Records;
46
48
public relationships: Relationships;
47
49
public interactions: Interactions;
···
71
73
this.moderation = new Moderation(db);
72
74
this.actors = new Actors(db);
73
75
this.identity = new Identity(idResolver);
76
+
this.notifications = new Notifications(db);
74
77
this.records = new Records(db);
75
78
this.relationships = new Relationships(db);
76
79
this.interactions = new Interactions(db);
+43
-5
data-plane/indexing/processor.ts
+43
-5
data-plane/indexing/processor.ts
···
4
4
import { lexicons } from "../../lex/lexicons.ts";
5
5
import { BackgroundQueue } from "../background.ts";
6
6
import { Database } from "../db/index.ts";
7
+
import { chunkArray } from "@atp/common";
7
8
8
9
// @NOTE re: insertions and deletions. Due to how record updates are handled,
9
10
// (insertFn) should have the same effect as (insertFn -> deleteFn -> insertFn).
···
292
293
}
293
294
}
294
295
295
-
handleNotifs(op: { deleted?: S; inserted?: S }) {
296
-
let _notifs: Notif[] = [];
296
+
async handleNotifs(op: { deleted?: S; inserted?: S }) {
297
+
let notifs: Notif[] = [];
298
+
const runOnCommit: ((db: Database) => Promise<void>)[] = [];
297
299
if (op.deleted) {
298
300
const forDelete = this.params.notifsForDelete(
299
301
op.deleted,
300
302
op.inserted ?? null,
301
303
);
302
-
_notifs = forDelete.notifs;
304
+
if (forDelete.toDelete.length > 0) {
305
+
// Notifs can be deleted in background: they are expensive to delete and
306
+
// listNotifications already excludes notifs with missing records.
307
+
runOnCommit.push(async (db) => {
308
+
await db.models.Notification.deleteMany({
309
+
recordUri: { $in: forDelete.toDelete },
310
+
});
311
+
});
312
+
}
313
+
notifs = forDelete.notifs;
303
314
} else if (op.inserted) {
304
-
_notifs = this.params.notifsForInsert(op.inserted);
315
+
notifs = this.params.notifsForInsert(op.inserted);
316
+
}
317
+
for (const chunk of chunkArray(notifs, 500)) {
318
+
runOnCommit.push(async (db) => {
319
+
const filtered = await this.filterNotifsForThreadMutes(chunk);
320
+
if (filtered.length > 0) {
321
+
await db.models.Notification.insertMany(
322
+
filtered.map((n) => ({
323
+
did: n.did,
324
+
recordUri: n.recordUri,
325
+
recordCid: n.recordCid,
326
+
author: n.author,
327
+
reason: n.reason,
328
+
reasonSubject: n.reasonSubject ?? null,
329
+
sortAt: n.sortAt,
330
+
})),
331
+
);
332
+
}
333
+
});
334
+
}
335
+
// Need to ensure notif deletion always happens before creation, otherwise delete may clobber in a race.
336
+
for (const fn of runOnCommit) {
337
+
await fn(this.appDb); // these could be backgrounded
305
338
}
339
+
}
306
340
307
-
// TODO: Implement notification handling
341
+
// Filter notifications for thread mutes (placeholder for future implementation)
342
+
filterNotifsForThreadMutes(notifs: Notif[]): Promise<Notif[]> {
343
+
// TODO: Implement thread mute filtering
344
+
// For now, return all notifications unfiltered
345
+
return Promise.resolve(notifs);
308
346
}
309
347
310
348
aggregateOnCommit(indexed: S) {
+239
data-plane/routes/notifs.ts
+239
data-plane/routes/notifs.ts
···
1
+
import { Database } from "../db/index.ts";
2
+
import { GenericKeyset } from "../db/pagination.ts";
3
+
4
+
type SortAtCidResult = { sortAt: string; recordCid: string };
5
+
type SortAtCidLabeledResult = { primary: string; secondary: string };
6
+
7
+
class SortAtCidKeyset extends GenericKeyset<
8
+
SortAtCidResult,
9
+
SortAtCidLabeledResult
10
+
> {
11
+
constructor() {
12
+
super("sortAt", "recordCid");
13
+
}
14
+
15
+
labelResult(result: SortAtCidResult): SortAtCidLabeledResult {
16
+
const sortAt = result.sortAt || new Date().toISOString();
17
+
return { primary: sortAt, secondary: result.recordCid };
18
+
}
19
+
20
+
labeledResultToCursor(labeled: SortAtCidLabeledResult) {
21
+
const timestamp = new Date(labeled.primary).getTime();
22
+
if (isNaN(timestamp)) {
23
+
throw new Error("Invalid date for cursor");
24
+
}
25
+
const secondsBase36 = Math.floor(timestamp / 1000).toString(36);
26
+
return {
27
+
primary: secondsBase36,
28
+
secondary: labeled.secondary,
29
+
};
30
+
}
31
+
32
+
cursorToLabeledResult(cursor: { primary: string; secondary: string }) {
33
+
const seconds = parseInt(cursor.primary, 36);
34
+
if (isNaN(seconds)) {
35
+
throw new Error("Malformed cursor: invalid timestamp");
36
+
}
37
+
const primaryDate = new Date(seconds * 1000);
38
+
if (isNaN(primaryDate.getTime())) {
39
+
throw new Error("Malformed cursor: invalid date");
40
+
}
41
+
return {
42
+
primary: primaryDate.toISOString(),
43
+
secondary: cursor.secondary,
44
+
};
45
+
}
46
+
}
47
+
48
+
export interface Notification {
49
+
recipientDid: string;
50
+
uri: string;
51
+
cid: string;
52
+
reason: string;
53
+
reasonSubject?: string;
54
+
sortAt: string;
55
+
authorDid: string;
56
+
priority?: boolean;
57
+
}
58
+
59
+
export class Notifications {
60
+
private db: Database;
61
+
private sortAtCidKeyset: SortAtCidKeyset;
62
+
63
+
constructor(db: Database) {
64
+
this.db = db;
65
+
this.sortAtCidKeyset = new SortAtCidKeyset();
66
+
}
67
+
68
+
async getNotifications(
69
+
actorDid: string,
70
+
limit = 50,
71
+
cursor?: string,
72
+
priority?: boolean,
73
+
): Promise<{ notifications: Notification[]; cursor?: string }> {
74
+
// Get follows for priority filtering
75
+
let priorityDids: string[] | undefined;
76
+
if (priority) {
77
+
const follows = await this.db.models.Follow.find({
78
+
authorDid: actorDid,
79
+
}).select("subject");
80
+
priorityDids = follows.map((f) => f.subject);
81
+
if (priorityDids.length === 0) {
82
+
return { notifications: [], cursor: undefined };
83
+
}
84
+
}
85
+
86
+
// Build base query
87
+
const baseFilter: Record<string, unknown> = { did: actorDid };
88
+
89
+
// If priority, filter to only notifications from followed users
90
+
if (priorityDids) {
91
+
baseFilter.author = { $in: priorityDids };
92
+
}
93
+
94
+
// Get notifications
95
+
const notifsQuery = this.db.models.Notification.find(baseFilter);
96
+
97
+
// Apply pagination
98
+
const paginatedQuery = this.sortAtCidKeyset.paginate(notifsQuery, {
99
+
limit,
100
+
cursor,
101
+
direction: "desc",
102
+
});
103
+
104
+
const notifs = await paginatedQuery.exec();
105
+
106
+
// Filter out notifications with missing reasonSubject records
107
+
const filteredNotifs = await this.filterValidReasonSubjects(notifs);
108
+
109
+
// Get priority status for each notification
110
+
const followedDids = priorityDids ?? await this.getFollowedDids(actorDid);
111
+
const followedSet = new Set(followedDids);
112
+
113
+
// Generate cursor from the last item if we have results
114
+
let nextCursor: string | undefined;
115
+
if (notifs.length === limit && notifs.length > 0) {
116
+
const lastNotif = notifs[notifs.length - 1];
117
+
nextCursor = this.sortAtCidKeyset.pack({
118
+
primary: lastNotif.sortAt,
119
+
secondary: lastNotif.recordCid,
120
+
});
121
+
}
122
+
123
+
const notifications = filteredNotifs.map((notif) => ({
124
+
recipientDid: actorDid,
125
+
uri: notif.recordUri,
126
+
cid: notif.recordCid,
127
+
reason: notif.reason,
128
+
reasonSubject: notif.reasonSubject ?? undefined,
129
+
sortAt: notif.sortAt,
130
+
authorDid: notif.author,
131
+
priority: followedSet.has(notif.author),
132
+
}));
133
+
134
+
return {
135
+
notifications,
136
+
cursor: nextCursor,
137
+
};
138
+
}
139
+
140
+
async getNotificationSeen(
141
+
actorDid: string,
142
+
_priority?: boolean,
143
+
): Promise<{ timestamp?: string }> {
144
+
const actor = await this.db.models.Actor.findOne({ did: actorDid });
145
+
if (!actor) {
146
+
return {};
147
+
}
148
+
149
+
// For now, we don't have lastSeenNotifs on Actor model
150
+
// This would need to be added to track notification seen status
151
+
// Returning empty for now
152
+
return {};
153
+
}
154
+
155
+
async getUnreadNotificationCount(
156
+
actorDid: string,
157
+
lastSeen?: string,
158
+
priority?: boolean,
159
+
): Promise<{ count: number }> {
160
+
const baseFilter: Record<string, unknown> = { did: actorDid };
161
+
162
+
// Filter by lastSeen if provided
163
+
if (lastSeen) {
164
+
baseFilter.sortAt = { $gt: lastSeen };
165
+
}
166
+
167
+
// If priority, filter to only notifications from followed users
168
+
if (priority) {
169
+
const follows = await this.db.models.Follow.find({
170
+
authorDid: actorDid,
171
+
}).select("subject");
172
+
const priorityDids = follows.map((f) => f.subject);
173
+
if (priorityDids.length === 0) {
174
+
return { count: 0 };
175
+
}
176
+
baseFilter.author = { $in: priorityDids };
177
+
}
178
+
179
+
const count = await this.db.models.Notification.countDocuments(baseFilter);
180
+
181
+
return { count };
182
+
}
183
+
184
+
async updateNotificationSeen(
185
+
_actorDid: string,
186
+
_timestamp: string,
187
+
_priority?: boolean,
188
+
): Promise<void> {
189
+
// This would require adding notification seen tracking to the Actor model
190
+
// or creating a separate ActorState model
191
+
// For now, this is a no-op
192
+
}
193
+
194
+
// Helper methods
195
+
196
+
private async getFollowedDids(actorDid: string): Promise<string[]> {
197
+
const follows = await this.db.models.Follow.find({
198
+
authorDid: actorDid,
199
+
}).select("subject");
200
+
return follows.map((f) => f.subject);
201
+
}
202
+
203
+
private async filterValidReasonSubjects(
204
+
notifs: Array<{
205
+
recordUri: string;
206
+
recordCid: string;
207
+
author: string;
208
+
reason: string;
209
+
reasonSubject: string | null;
210
+
sortAt: string;
211
+
}>,
212
+
): Promise<
213
+
Array<{
214
+
recordUri: string;
215
+
recordCid: string;
216
+
author: string;
217
+
reason: string;
218
+
reasonSubject: string | null;
219
+
sortAt: string;
220
+
}>
221
+
> {
222
+
// Filter out notifications where reasonSubject exists but the record doesn't
223
+
const notifsWithSubject = notifs.filter((n) => n.reasonSubject);
224
+
if (notifsWithSubject.length === 0) {
225
+
return notifs;
226
+
}
227
+
228
+
const subjectUris = notifsWithSubject.map((n) => n.reasonSubject as string);
229
+
const existingRecords = await this.db.models.Record.find({
230
+
uri: { $in: subjectUris },
231
+
}).select("uri");
232
+
233
+
const existingUris = new Set(existingRecords.map((r) => r.uri));
234
+
235
+
return notifs.filter(
236
+
(n) => !n.reasonSubject || existingUris.has(n.reasonSubject),
237
+
);
238
+
}
239
+
}
+66
-1
hydration/index.ts
+66
-1
hydration/index.ts
···
41
41
GraphHydrator,
42
42
RelationshipPair,
43
43
} from "./graph.ts";
44
-
import { HydrationMap, ItemRef, mergeMaps, RecordInfo } from "./util.ts";
44
+
import {
45
+
HydrationMap,
46
+
ItemRef,
47
+
mergeMaps,
48
+
RecordInfo,
49
+
urisByCollection,
50
+
} from "./util.ts";
45
51
import { getLogger } from "@logtape/logtape";
46
52
import {
47
53
LabelerAggs,
···
51
57
Labels,
52
58
} from "./label.ts";
53
59
import { ParsedLabelers } from "../util.ts";
60
+
import { Notification } from "../data-plane/routes/notifs.ts";
54
61
55
62
export class HydrateCtx {
56
63
labelers: ParsedLabelers;
···
644
651
this.hydrateProfiles(uris.map(didFromUri), ctx),
645
652
]);
646
653
return mergeStates(profileState, { reposts, ctx });
654
+
}
655
+
656
+
// so.sprk.notification.listNotifications#notification
657
+
// - notification
658
+
// - profile
659
+
// - list basic`
660
+
async hydrateNotifications(
661
+
notifs: Notification[],
662
+
ctx: HydrateCtx,
663
+
): Promise<HydrationState> {
664
+
const uris = notifs.map((notif) => notif.uri);
665
+
const collections = urisByCollection(uris);
666
+
const postUris = collections.get(ids.SoSprkFeedPost) ?? [];
667
+
const replyUris = collections.get(ids.SoSprkFeedReply) ?? [];
668
+
const likeUris = collections.get(ids.SoSprkFeedLike) ?? [];
669
+
const repostUris = collections.get(ids.SoSprkFeedRepost) ?? [];
670
+
const followUris = collections.get(ids.SoSprkGraphFollow) ?? [];
671
+
const [
672
+
posts,
673
+
replies,
674
+
likes,
675
+
reposts,
676
+
follows,
677
+
labels,
678
+
profileState,
679
+
] = await Promise.all([
680
+
this.feed.getPosts(postUris), // reason: mention, quote
681
+
this.feed.getReplies(replyUris), // reason: reply
682
+
this.feed.getLikes(likeUris), // reason: like
683
+
this.feed.getReposts(repostUris), // reason: repost
684
+
this.graph.getFollows(followUris), // reason: follow
685
+
this.label.getLabelsForSubjects(uris, ctx.labelers),
686
+
this.hydrateProfiles(uris.map(didFromUri), ctx),
687
+
]);
688
+
const viewerRootPostUris = new Set<string>();
689
+
for (const notif of notifs) {
690
+
if (notif.reason === "reply") {
691
+
// Check replies map for reply notifications
692
+
const reply = replies.get(notif.uri);
693
+
if (reply) {
694
+
const rootUri = reply.record.reply?.root.uri;
695
+
if (rootUri && didFromUri(rootUri) === ctx.viewer) {
696
+
viewerRootPostUris.add(rootUri);
697
+
}
698
+
}
699
+
}
700
+
}
701
+
actionTakedownLabels(postUris, posts, labels);
702
+
actionTakedownLabels(replyUris, replies, labels);
703
+
return mergeStates(profileState, {
704
+
posts,
705
+
replies,
706
+
likes,
707
+
reposts,
708
+
follows,
709
+
labels,
710
+
ctx,
711
+
});
647
712
}
648
713
649
714
// so.sprk.sound.defs#audioView
+96
-1
views/index.ts
+96
-1
views/index.ts
···
48
48
Media,
49
49
MediaView,
50
50
NotFoundPost,
51
+
NotificationView,
51
52
VideoMedia,
52
53
VideoMediaView,
53
54
} from "./types.ts";
···
66
67
import { cidFromBlobJson } from "./util.ts";
67
68
import { uriToDid } from "../utils/uris.ts";
68
69
import { mapDefined } from "@atp/common";
69
-
import { FeedItem } from "../hydration/feed.ts";
70
+
import { FeedItem, Like, Post, Reply, Repost } from "../hydration/feed.ts";
70
71
import {
71
72
QueryParams as GetThreadQueryParams,
72
73
ThreadItem,
···
82
83
LabelerViewDetailed,
83
84
} from "../lex/types/so/sprk/labeler/defs.ts";
84
85
import { isSelfLabels } from "../lex/types/com/atproto/label/defs.ts";
86
+
import { Follow } from "../hydration/graph.ts";
87
+
import { RecordInfo } from "../hydration/util.ts";
88
+
import { Notification } from "../data-plane/routes/notifs.ts";
85
89
86
90
export class Views {
87
91
public indexedAtEpoch?: Date | undefined;
···
1030
1034
if (actor?.upstreamStatus === "takendown") return true;
1031
1035
if (actor?.upstreamStatus === "suspended") return true;
1032
1036
return false;
1037
+
}
1038
+
1039
+
viewerSeesNeedsReview(
1040
+
{ did, uri }: { did?: string; uri?: string },
1041
+
state: HydrationState,
1042
+
): boolean {
1043
+
const { labels, profileViewers, ctx } = state;
1044
+
did = did || (uri && uriToDid(uri));
1045
+
if (!did) {
1046
+
return true;
1047
+
}
1048
+
if (
1049
+
labels?.get(did)?.needsReview ||
1050
+
(uri && labels?.get(uri)?.needsReview)
1051
+
) {
1052
+
// content marked as needs review
1053
+
return ctx?.viewer === did || !!profileViewers?.get(did)?.following;
1054
+
}
1055
+
return true;
1056
+
}
1057
+
1058
+
notification(
1059
+
notif: Notification,
1060
+
lastSeenAt: string | undefined,
1061
+
state: HydrationState,
1062
+
): Un$Typed<NotificationView> | undefined {
1063
+
if (!notif.sortAt || !notif.reason) return;
1064
+
const uri = new AtUri(notif.uri);
1065
+
const authorDid = notif.authorDid;
1066
+
const author = this.profile(authorDid, state);
1067
+
if (!author) return;
1068
+
1069
+
let recordInfo:
1070
+
| Post
1071
+
| Reply
1072
+
| Like
1073
+
| Repost
1074
+
| Follow
1075
+
| RecordInfo<ProfileRecord>
1076
+
| undefined
1077
+
| null;
1078
+
1079
+
if (uri.collection === ids.SoSprkFeedPost) {
1080
+
recordInfo = state.posts?.get(notif.uri);
1081
+
} else if (uri.collection === ids.SoSprkFeedReply) {
1082
+
recordInfo = state.replies?.get(notif.uri);
1083
+
} else if (uri.collection === ids.SoSprkFeedLike) {
1084
+
recordInfo = state.likes?.get(notif.uri);
1085
+
} else if (uri.collection === ids.SoSprkFeedRepost) {
1086
+
recordInfo = state.reposts?.get(notif.uri);
1087
+
} else if (uri.collection === ids.SoSprkGraphFollow) {
1088
+
recordInfo = state.follows?.get(notif.uri);
1089
+
} else if (uri.collection === ids.SoSprkActorProfile) {
1090
+
const actor = state.actors?.get(authorDid);
1091
+
recordInfo = actor && actor.profile && actor.profileCid
1092
+
? {
1093
+
record: actor.profile,
1094
+
cid: actor.profileCid,
1095
+
sortedAt: actor.sortedAt ?? new Date(0), // @NOTE will be present since profile record is present
1096
+
indexedAt: actor.indexedAt ?? new Date(0), // @NOTE will be present since profile record is present
1097
+
takedownRef: actor.profileTakedownRef,
1098
+
}
1099
+
: undefined;
1100
+
}
1101
+
if (!recordInfo) return;
1102
+
1103
+
const labels = state.labels?.getBySubject(notif.uri) ?? [];
1104
+
// selfLabels only applies to posts and profiles, safe to pass the record
1105
+
const selfLabels = isPostRecord(recordInfo.record) ||
1106
+
isProfileRecord(recordInfo.record)
1107
+
? this.selfLabels({
1108
+
uri: notif.uri,
1109
+
cid: recordInfo.cid,
1110
+
record: recordInfo.record,
1111
+
})
1112
+
: [];
1113
+
const indexedAt = notif.sortAt;
1114
+
return {
1115
+
uri: notif.uri,
1116
+
cid: recordInfo.cid,
1117
+
author,
1118
+
reason: notif.reason as NotificationView["reason"],
1119
+
reasonSubject: notif.reasonSubject || undefined,
1120
+
record: recordInfo.record,
1121
+
// @NOTE works with a hack in listNotifications so that when there's no last-seen time,
1122
+
// the user's first notification is marked unread, and all previous read. in this case,
1123
+
// the last seen time will be equal to the first notification's indexed time.
1124
+
isRead: lastSeenAt ? lastSeenAt > indexedAt : true,
1125
+
indexedAt: notif.sortAt,
1126
+
labels: [...labels, ...selfLabels],
1127
+
};
1033
1128
}
1034
1129
}
+4
views/types.ts
+4
views/types.ts
···
15
15
} from "../lex/types/so/sprk/feed/defs.ts";
16
16
import { LabelerView } from "../lex/types/so/sprk/labeler/defs.ts";
17
17
18
+
export type {
19
+
Notification as NotificationView,
20
+
} from "../lex/types/so/sprk/notification/listNotifications.ts";
21
+
18
22
export {
19
23
isMain as isImagesMedia,
20
24
type Main as ImagesMedia,