whey.party
an attempt to make a lightweight, easily self-hostable, scoped bluesky appview
initial index server instantiation
+1
-1
.env
+1
-1
.env
+5
-2
index/types.ts
+5
-2
index/types.ts
···
1
+
import { Database } from "jsr:@db/sqlite@0.11";
2
+
1
3
export type indexHandlerContext = {
2
4
op: string;
3
5
doer: string; // the formal term for this is "repo" but whatever right
···
5
7
cid?: string;
6
8
aturi: string;
7
9
indexsrc: string;
8
-
value: Record<string, unknown>
9
-
userdbname: string;
10
+
value: Record<string, unknown>;
11
+
//userdbname: string;
12
+
db: Database;
10
13
}
+299
-3
indexserver.ts
+299
-3
indexserver.ts
···
3
3
import { validateRecord } from "./utils/records.ts";
4
4
import { searchParamsToJson, withCors } from "./utils/server.ts";
5
5
import * as IndexServerTypes from "./utils/indexservertypes.ts";
6
+
import { Database } from "jsr:@db/sqlite@0.11";
7
+
import { setupUserDb } from "./utils/dbuser.ts";
8
+
import { jetstreamurl } from "./main.ts";
9
+
import { JetstreamManager, SpacedustManager } from "./utils/sharders.ts";
10
+
import { handleSpacedust, SpacedustLinkMessage } from "./index/spacedust.ts";
11
+
import { handleJetstream } from "./index/jetstream.ts";
12
+
import { AtUri } from "npm:@atproto/api";
13
+
14
+
export class IndexServerUserManager {
15
+
private users = new Map<string, UserIndexServer>();
16
+
17
+
/*async*/ addUser(did: string) {
18
+
if (this.users.has(did)) return;
19
+
const instance = new UserIndexServer(did);
20
+
//await instance.initialize();
21
+
this.users.set(did, instance);
22
+
}
23
+
24
+
// async handleRequest({
25
+
// did,
26
+
// route,
27
+
// req,
28
+
// }: {
29
+
// did: string;
30
+
// route: string;
31
+
// req: Request;
32
+
// }) {
33
+
// if (!this.users.has(did)) await this.addUser(did);
34
+
// const user = this.users.get(did)!;
35
+
// return await user.handleHttpRequest(route, req);
36
+
// }
37
+
38
+
removeUser(did: string) {
39
+
const instance = this.users.get(did);
40
+
if (!instance) return;
41
+
/*await*/ instance.shutdown();
42
+
this.users.delete(did);
43
+
}
44
+
45
+
getDbForDid(did: string): Database | null {
46
+
if (!this.users.has(did)) {
47
+
return null
48
+
}
49
+
return this.users.get(did)?.db ?? null;
50
+
}
51
+
52
+
coldStart(db: Database) {
53
+
const rows = db.prepare("SELECT did FROM users").all();
54
+
for (const row of rows) {
55
+
this.addUser(row.did);
56
+
}
57
+
}
58
+
}
59
+
60
+
class UserIndexServer {
61
+
did: string;
62
+
db: Database;// | undefined;
63
+
jetstream: JetstreamManager;// | undefined;
64
+
spacedust: SpacedustManager;// | undefined;
65
+
66
+
constructor(did: string) {
67
+
this.did = did;
68
+
this.db = openDbForDid(this.did);
69
+
// should probably put the params of exactly what were listening to here
70
+
this.jetstream = new JetstreamManager((msg) => {
71
+
console.log("Received Jetstream message: ", msg);
72
+
73
+
const op = msg.commit.operation;
74
+
const doer = msg.did;
75
+
const rev = msg.commit.rev;
76
+
const aturi = `${msg.did}/${msg.commit.collection}/${msg.commit.rkey}`;
77
+
const value = msg.commit.record;
78
+
79
+
if (!doer || !value) return;
80
+
indexServerIndexer({
81
+
op,
82
+
doer,
83
+
rev,
84
+
aturi,
85
+
value,
86
+
indexsrc: `jetstream-${op}`,
87
+
db: this.db,
88
+
});
89
+
});
90
+
this.jetstream.start({
91
+
// for realsies pls get from db or something instead of this shit
92
+
wantedDids: [
93
+
this.did
94
+
// "did:plc:mn45tewwnse5btfftvd3powc",
95
+
// "did:plc:yy6kbriyxtimkjqonqatv2rb",
96
+
// "did:plc:zzhzjga3ab5fcs2vnsv2ist3",
97
+
// "did:plc:jz4ibztn56hygfld6j6zjszg",
98
+
],
99
+
wantedCollections: [
100
+
"app.bsky.actor.profile",
101
+
"app.bsky.feed.generator",
102
+
"app.bsky.feed.like",
103
+
"app.bsky.feed.post",
104
+
"app.bsky.feed.repost",
105
+
"app.bsky.feed.threadgate",
106
+
"app.bsky.graph.block",
107
+
"app.bsky.graph.follow",
108
+
"app.bsky.graph.list",
109
+
"app.bsky.graph.listblock",
110
+
"app.bsky.graph.listitem",
111
+
"app.bsky.notification.declaration",
112
+
],
113
+
});
114
+
//await connectToJetstream(this.did, this.db);
115
+
this.spacedust = new SpacedustManager((msg: SpacedustLinkMessage) => {
116
+
console.log("Received Spacedust message: ", msg);
117
+
const sourceURI = new AtUri(msg.link.source_record);
118
+
const srcUri = msg.link.source_record
119
+
const srcDid = sourceURI.host
120
+
const srcField = msg.link.source
121
+
const srcCol = sourceURI.collection
122
+
const subjectURI = new AtUri(msg.link.subject)
123
+
const subUri = msg.link.subject
124
+
const subDid = subjectURI.host
125
+
const subCol = subjectURI.collection
126
+
127
+
this.db.run(
128
+
`INSERT INTO backlink_skeleton (
129
+
srcuri,
130
+
srcdid,
131
+
srcfield,
132
+
srccol,
133
+
suburi,
134
+
subdid,
135
+
subcol
136
+
) VALUES (?, ?, ?, ?, ?, ?, ?)`,
137
+
[
138
+
srcUri, // full AT URI of the source record
139
+
srcDid, // did: of the source
140
+
srcField, // e.g., "reply.parent.uri" or "facets.features.did"
141
+
srcCol, // e.g., "app.bsky.feed.post"
142
+
subUri, // full AT URI of the subject (linked record)
143
+
subDid, // did: of the subject
144
+
subCol, // subject collection (can be inferred or passed)
145
+
]
146
+
);
147
+
});
148
+
this.spacedust.start({
149
+
wantedSources: [
150
+
"app.bsky.feed.like:subject.uri", // like
151
+
"app.bsky.feed.like:via.uri", // liked repost
152
+
"app.bsky.feed.repost:subject.uri", // repost
153
+
"app.bsky.feed.repost:via.uri", // reposted repost
154
+
"app.bsky.feed.post:reply.root.uri", // thread OP
155
+
"app.bsky.feed.post:reply.parent.uri", // direct parent
156
+
"app.bsky.feed.post:embed.media.record.record.uri", // quote with media
157
+
"app.bsky.feed.post:embed.record.uri", // quote without media
158
+
"app.bsky.feed.threadgate:post", // threadgate subject
159
+
"app.bsky.feed.threadgate:hiddenReplies", // threadgate items (array)
160
+
"app.bsky.feed.post:facets.features.did", // facet item (array): mention
161
+
"app.bsky.graph.block:subject", // blocks
162
+
"app.bsky.graph.follow:subject", // follow
163
+
"app.bsky.graph.listblock:subject", // list item (blocks)
164
+
"app.bsky.graph.listblock:list", // blocklist mention (might not exist)
165
+
"app.bsky.graph.listitem:subject", // list item (blocks)
166
+
"app.bsky.graph.listitem:list", // list mention
167
+
],
168
+
// should be getting from DB but whatever right
169
+
wantedSubjects: [
170
+
// as noted i dont need to write down each post, just the user to listen to !
171
+
// hell yeah
172
+
// "at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3lvybv7b6ic2h",
173
+
// "at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3lvybws4avc2h",
174
+
// "at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3lvvkcxcscs2h",
175
+
// "at://did:plc:yy6kbriyxtimkjqonqatv2rb/app.bsky.feed.post/3l63ogxocq42f",
176
+
// "at://did:plc:yy6kbriyxtimkjqonqatv2rb/app.bsky.feed.post/3lw3wamvflu23",
177
+
],
178
+
wantedSubjectDids: [
179
+
this.did,
180
+
//"did:plc:mn45tewwnse5btfftvd3powc",
181
+
//"did:plc:yy6kbriyxtimkjqonqatv2rb",
182
+
//"did:plc:zzhzjga3ab5fcs2vnsv2ist3",
183
+
//"did:plc:jz4ibztn56hygfld6j6zjszg",
184
+
],
185
+
});
186
+
//await connectToConstellation(this.did, this.db);
187
+
}
188
+
189
+
// initialize() {
190
+
191
+
// }
192
+
193
+
// async handleHttpRequest(route: string, req: Request): Promise<Response> {
194
+
// if (route === "posts") {
195
+
// const posts = await this.queryPosts();
196
+
// return new Response(JSON.stringify(posts), {
197
+
// headers: { "content-type": "application/json" },
198
+
// });
199
+
// }
200
+
201
+
// return new Response("Unknown route", { status: 404 });
202
+
// }
203
+
204
+
// private async queryPosts() {
205
+
// return this.db.run(
206
+
// "SELECT * FROM posts ORDER BY created_at DESC LIMIT 100"
207
+
// );
208
+
// }
209
+
210
+
shutdown() {
211
+
this.jetstream.stop();
212
+
this.spacedust.stop();
213
+
this.db.close?.();
214
+
}
215
+
}
216
+
217
+
function openDbForDid(did: string): Database {
218
+
const path = `./dbs/${did}.sqlite`;
219
+
const db = new Database(path);
220
+
setupUserDb(db);
221
+
//await db.exec(/* CREATE IF NOT EXISTS statements */);
222
+
return db;
223
+
}
224
+
225
+
// async function connectToJetstream(did: string, db: Database): Promise<WebSocket> {
226
+
// const url = `${jetstreamurl}/xrpc/com.atproto.sync.subscribeRepos?did=${did}`;
227
+
// const ws = new WebSocket(url);
228
+
// ws.onmessage = (msg) => {
229
+
// //handleJetstreamMessage(evt.data, db)
230
+
231
+
// const op = msg.commit.operation;
232
+
// const doer = msg.did;
233
+
// const rev = msg.commit.rev;
234
+
// const aturi = `${msg.did}/${msg.commit.collection}/${msg.commit.rkey}`;
235
+
// const value = msg.commit.record;
236
+
237
+
// if (!doer || !value) return;
238
+
// indexServerIndexer({
239
+
// op,
240
+
// doer,
241
+
// rev,
242
+
// aturi,
243
+
// value,
244
+
// indexsrc: "onboarding_backfill",
245
+
// userdbname: did,
246
+
// })
247
+
// };
248
+
249
+
// return ws;
250
+
// }
251
+
252
+
// async function connectToConstellation(did: string, db: D1Database): Promise<WebSocket> {
253
+
// const url = `wss://bsky.social/xrpc/com.atproto.sync.subscribeLabels?did=${did}`;
254
+
// const ws = new WebSocket(url);
255
+
// ws.onmessage = (evt) => handleConstellationMessage(evt.data, db);
256
+
// return ws;
257
+
// }
6
258
7
259
export async function indexServerHandler(req: Request): Promise<Response> {
8
260
const url = new URL(req.url);
···
221
473
target: string;
222
474
};
223
475
224
-
export async function constellationAPIHandler(req: Request): Promise<Response> {
476
+
const SQL = {
477
+
links: `
478
+
SELECT srcuri, srcdid, srccol
479
+
FROM backlink_skeleton
480
+
WHERE suburi = ? AND subcol = ? AND srcfield = ?
481
+
`,
482
+
distinctDids: `
483
+
SELECT DISTINCT srcdid
484
+
FROM backlink_skeleton
485
+
WHERE suburi = ? AND subcol = ? AND srcfield = ?
486
+
`,
487
+
count: `
488
+
SELECT COUNT(*) as total
489
+
FROM backlink_skeleton
490
+
WHERE suburi = ? AND subcol = ? AND srcfield = ?
491
+
`,
492
+
countDistinctDids: `
493
+
SELECT COUNT(DISTINCT srcdid) as total
494
+
FROM backlink_skeleton
495
+
WHERE suburi = ? AND subcol = ? AND srcfield = ?
496
+
`,
497
+
all: `
498
+
SELECT suburi, srccol, COUNT(*) as records, COUNT(DISTINCT srcdid) as distinct_dids
499
+
FROM backlink_skeleton
500
+
WHERE suburi = ?
501
+
GROUP BY suburi, srccol
502
+
`,
503
+
}
504
+
505
+
export async function constellationAPIHandler(req: Request, did: string): Promise<Response> {
225
506
const url = new URL(req.url);
226
507
const pathname = url.pathname;
227
508
// const bskyUrl = `https://api.bsky.app${pathname}${url.search}`;
···
231
512
// : null;
232
513
const searchParams = searchParamsToJson(url.searchParams);
233
514
const jsonUntyped = searchParams;
515
+
const db = openDbForDid(did);
234
516
235
517
switch (pathname) {
236
518
case "/links": {
237
519
const jsonTyped = jsonUntyped as linksQuery;
520
+
// probably need to do pagination or something
238
521
239
-
const response: linksRecordsResponse = {};
522
+
const rows = db.prepare(SQL.links).all(jsonTyped.target, jsonTyped.collection, jsonTyped.path);
523
+
524
+
const linking_records: linksRecord[] = rows.map((row: any) => {
525
+
const rkey = row.srcuri.split('/').pop()!;
526
+
return {
527
+
did: row.srcdid,
528
+
collection: row.srccol,
529
+
rkey,
530
+
};
531
+
});
240
532
533
+
const response: linksRecordsResponse = {
534
+
total: linking_records.length.toString(),
535
+
linking_records,
536
+
};
241
537
return new Response(JSON.stringify(response), {
242
538
headers: withCors({ "Content-Type": "application/json" }),
243
539
});
···
283
579
JSON.stringify({
284
580
error: "idk NotSupported",
285
581
message:
286
-
"HEY hello there my name is whey dot party and you have used my custom appview that is very cool but have you considered that idk Not Supported",
582
+
"HEY hello there my name is whey dot party and you have used my custom constellation implementation that is very cool but have you considered that idk Not Supported",
287
583
}),
288
584
{
289
585
status: 404,
+5
-1
main.ts
+5
-1
main.ts
···
14
14
import * as ATPAPI from "npm:@atproto/api";
15
15
import { didDocument } from "./utils/diddoc.ts";
16
16
import { cachedFetch, searchParamsToJson, withCors } from "./utils/server.ts";
17
-
import { constellationAPIHandler, indexServerHandler } from "./indexserver.ts";
17
+
import { constellationAPIHandler, indexServerHandler, IndexServerUserManager } from "./indexserver.ts";
18
18
import { viewServerHandler } from "./viewserver.ts";
19
19
20
+
export const jetstreamurl = Deno.env.get("JETSTREAM_URL");
20
21
export const slingshoturl = Deno.env.get("SLINGSHOT_URL");
21
22
export const constellationurl = Deno.env.get("CONSTELLATION_URL");
22
23
export const spacedusturl = Deno.env.get("SPACEDUST_URL");
···
27
28
28
29
export const systemDB = new Database("system.db");
29
30
setupSystemDb(systemDB);
31
+
32
+
const userManager = new IndexServerUserManager();
33
+
userManager.coldStart(systemDB)
30
34
31
35
// should do both of these per user actually, since now each user has their own db
32
36
// also the set of records and backlinks to listen should be seperate between index and view servers