graham.systems
forked from
samuel.fm/statusphere-react
the statusphere demo reworked into a vite/react app in a monorepo
Fix: handle firehose update events
+6
-1
src/firehose/firehose.ts
+6
-1
src/firehose/firehose.ts
···
57
if (isCommit(evt) && !this.opts.excludeCommit) {
58
const parsed = await parseCommit(evt)
59
for (const write of parsed) {
60
-
if (!this.opts.filterCollections || this.opts.filterCollections.includes(write.uri.collection)) {
61
yield write
62
}
63
}
···
167
168
type Update = CommitMeta & {
169
event: 'update'
170
}
171
172
type Delete = CommitMeta & {
···
57
if (isCommit(evt) && !this.opts.excludeCommit) {
58
const parsed = await parseCommit(evt)
59
for (const write of parsed) {
60
+
if (
61
+
!this.opts.filterCollections ||
62
+
this.opts.filterCollections.includes(write.uri.collection)
63
+
) {
64
yield write
65
}
66
}
···
170
171
type Update = CommitMeta & {
172
event: 'update'
173
+
record: RepoRecord
174
+
cid: CID
175
}
176
177
type Delete = CommitMeta & {
+12
-2
src/firehose/ingester.ts
+12
-2
src/firehose/ingester.ts
···
10
const firehose = new Firehose({})
11
12
for await (const evt of firehose.run()) {
13
-
if (evt.event === 'create') {
14
const record = evt.record
15
if (
16
evt.collection === 'com.example.status' &&
17
Status.isRecord(record) &&
18
Status.validateRecord(record).success
19
) {
20
await this.db
21
.insertInto('status')
22
.values({
···
25
updatedAt: record.updatedAt,
26
indexedAt: new Date().toISOString(),
27
})
28
-
.onConflict((oc) => oc.doNothing())
29
.execute()
30
}
31
}
···
10
const firehose = new Firehose({})
11
12
for await (const evt of firehose.run()) {
13
+
// Watch for write events
14
+
if (evt.event === 'create' || evt.event === 'update') {
15
const record = evt.record
16
+
17
+
// If the write is a valid status update
18
if (
19
evt.collection === 'com.example.status' &&
20
Status.isRecord(record) &&
21
Status.validateRecord(record).success
22
) {
23
+
// Store the status in our SQLite
24
await this.db
25
.insertInto('status')
26
.values({
···
29
updatedAt: record.updatedAt,
30
indexedAt: new Date().toISOString(),
31
})
32
+
.onConflict((oc) =>
33
+
oc.column('authorDid').doUpdateSet({
34
+
status: record.status,
35
+
updatedAt: record.updatedAt,
36
+
indexedAt: new Date().toISOString(),
37
+
})
38
+
)
39
.execute()
40
}
41
}