this repo has no description
created an initial test case
+66
-4
+5
-3
apps/firehose/index.ts
+5
-3
apps/firehose/index.ts
···
1
-
import { JetstreamSubscription } from "@atcute/jetstream";
1
+
import { JetstreamSubscription, type JetstreamEvent } from "@atcute/jetstream";
2
2
import { is } from '@atcute/lexicons';
3
3
import { CaAnsxorCatnipTrack } from "lexicon/atcute-lexicon";
4
4
···
7
7
wantedCollections: ["ca.ansxor.catnip.track"]
8
8
});
9
9
10
-
async function loop(subscription: JetstreamSubscription) {
10
+
export async function loop(subscription: AsyncIterable<JetstreamEvent>) {
11
11
12
12
for await (const event of subscription) {
13
13
if (event.kind !== "commit") {
···
29
29
}
30
30
}
31
31
32
-
await loop(subscription);
32
+
if (import.meta.main) {
33
+
await loop(subscription);
34
+
}
+2
-1
apps/firehose/package.json
+2
-1
apps/firehose/package.json
+58
apps/firehose/tests/firehose.test.ts
+58
apps/firehose/tests/firehose.test.ts
···
1
1
import { test, expect, beforeAll, afterAll } from "bun:test";
2
2
import { setupTestDb, teardownTestDb } from "db/test-utils";
3
+
import type { InferOutput } from "@atcute/lexicons/validations";
4
+
import { loop } from "../index";
5
+
import type { JetstreamEvent } from "@atcute/jetstream";
6
+
import type { CaAnsxorCatnipTrack } from "lexicon/atcute-lexicon";
3
7
4
8
let db: Awaited<ReturnType<typeof setupTestDb>>;
5
9
···
11
15
await teardownTestDb();
12
16
});
13
17
18
+
function mockSubscription(events: JetstreamEvent[]): AsyncIterable<JetstreamEvent> {
19
+
return {
20
+
async *[Symbol.asyncIterator]() {
21
+
for (const event of events) {
22
+
yield event;
23
+
}
24
+
},
25
+
};
26
+
}
27
+
28
+
const SAMPLE_DID = "did:plc:ragtjsm2j2vknwkz3zp4oxrd";
29
+
const SAMPLE_RKEY = "3jui7kd54zh2i";
30
+
const SAMPLE_CID = "bafyreig2fjxi3a743bb54z2at6lsv2brz4f5d3hefma6ludqt5d2k2ofey";
31
+
const SAMPLE_COLLECTION = "ca.ansxor.catnip.track";
32
+
33
+
const createTrackEvent: JetstreamEvent = {
34
+
did: SAMPLE_DID,
35
+
time_us: Date.now() * 1000,
36
+
kind: "commit",
37
+
commit: {
38
+
operation: "create",
39
+
collection: SAMPLE_COLLECTION,
40
+
rkey: SAMPLE_RKEY,
41
+
rev: "3jui7kd54zh2i",
42
+
cid: SAMPLE_CID,
43
+
record: {
44
+
$type: "ca.ansxor.catnip.track",
45
+
title: "Midnight Echoes",
46
+
createdAt: new Date().toISOString(),
47
+
audio: {
48
+
$type: "blob",
49
+
ref: { $link: "bafyreig2fjxi3a743bb54z2at6lsv2brz4f5d3hefma6ludqt5d2k2ofey" },
50
+
mimeType: "audio/ogg",
51
+
size: 5_000_000,
52
+
},
53
+
} satisfies InferOutput<typeof CaAnsxorCatnipTrack.mainSchema>,
54
+
},
55
+
};
56
+
14
57
test("can query tracks", async () => {
15
58
const tracks = await db.query.tracks.findMany();
16
59
expect(tracks).toEqual([]);
17
60
});
61
+
62
+
test("inserts track from create commit event", async () => {
63
+
const subscription = mockSubscription([createTrackEvent]);
64
+
await loop(subscription);
65
+
66
+
const tracks = await db.query.tracks.findMany();
67
+
expect(tracks).toHaveLength(1);
68
+
const track = tracks[0];
69
+
expect(track).not.toBeUndefined();
70
+
expect(track!.uri).toBe(`at://${SAMPLE_DID}/${SAMPLE_COLLECTION}/${SAMPLE_RKEY}`);
71
+
expect(track!.did).toBe(SAMPLE_DID);
72
+
expect(track!.rkey).toBe(SAMPLE_RKEY);
73
+
expect(track!.cid).toBe(SAMPLE_CID);
74
+
expect(track!.title).toBe("Midnight Echoes");
75
+
});