graham.systems / statusphere-react
forked from samuel.fm/statusphere-react
the statusphere demo reworked into a vite/react app in a monorepo
fork atom

tidy firehose

dholms 9990f3e3 f99257fb

+2 -165
+2 -165
src/firehose.ts
··· 1 - // import { BlobRef } from "@atproto/lexicon"; 2 1 import { cborToLexRecord, readCar } from "@atproto/repo"; 3 2 import { Subscription } from "@atproto/xrpc-server"; 4 3 import { Post } from "#/db"; 5 - // import type { Database } from "../db"; 6 - // import { ids, lexicons } from "../lexicon/lexicons"; 7 - // import type { Record as LikeRecord } from "../lexicon/types/app/bsky/feed/like"; 8 - // import type { Record as PostRecord } from "../lexicon/types/app/bsky/feed/post"; 9 - // import type { Record as RepostRecord } from "../lexicon/types/app/bsky/feed/repost"; 10 - // import type { Record as FollowRecord } from "../lexicon/types/app/bsky/graph/follow"; 11 - // import { 12 - // type Commit, 13 - // type OutputSchema as RepoEvent, 14 - // isCommit, 15 - // } from "../lexicon/types/com/atproto/sync/subscribeRepos"; 16 4 17 5 export class Firehose { 18 6 public sub: Subscription<unknown>; ··· 21 9 this.sub = new Subscription({ 22 10 service: service, 23 11 method: "com.atproto.sync.subscribeRepos", 24 - getParams: () => this.getCursor(), 25 - validate: (value: unknown) => { 26 - return value; 27 - // try { 28 - // return lexicons.assertValidXrpcMessage<RepoEvent>( 29 - // ids.ComAtprotoSyncSubscribeRepos, 30 - // value 31 - // ); 32 - // } catch (err) { 33 - // console.error("repo subscription skipped invalid message", err); 34 - // } 35 - }, 12 + getParams: () => ({}), 13 + validate: (value: unknown) => value, 36 14 }); 37 15 } 38 16 ··· 65 43 } catch (err) { 66 44 console.error("repo subscription could not handle message", err); 67 45 } 68 - // // update stored cursor every 20 events or so 69 - // if (isCommit(evt) && evt.seq % 20 === 0) { 70 - // await this.updateCursor(evt.seq); 71 - // } 72 46 } 73 47 } catch (err) { 74 48 console.error("repo subscription errored", err); 75 49 setTimeout(() => this.run(subscriptionReconnectDelay), subscriptionReconnectDelay); 76 50 } 77 51 } 78 - 79 - async getCursor() { 80 - return {}; 81 - } 82 - 83 - // async updateCursor(cursor: number) { 84 - // await this.db 85 - // .updateTable("sub_state") 86 - // .set({ cursor }) 87 - // .where("service", "=", this.service) 88 - // .execute(); 89 - // } 90 - 91 - // async getCursor(): Promise<{ cursor?: number }> { 92 - // const res = await this.db 93 - // .selectFrom("sub_state") 94 - // .selectAll() 95 - // .where("service", "=", this.service) 96 - // .executeTakeFirst(); 97 - // return res ? { cursor: res.cursor } : {}; 98 - // } 99 52 } 100 - 101 - // export const getOpsByType = async (evt: Commit): Promise<OperationsByType> => { 102 - // const car = await readCar(evt.blocks); 103 - // const opsByType: OperationsByType = { 104 - // posts: { creates: [], deletes: [] }, 105 - // reposts: { creates: [], deletes: [] }, 106 - // likes: { creates: [], deletes: [] }, 107 - // follows: { creates: [], deletes: [] }, 108 - // }; 109 - 110 - // for (const op of evt.ops) { 111 - // const uri = `at://${evt.repo}/${op.path}`; 112 - // const [collection] = op.path.split("/"); 113 - 114 - // if (op.action === "update") continue; // updates not supported yet 115 - 116 - // if (op.action === "create") { 117 - // if (!op.cid) continue; 118 - // const recordBytes = car.blocks.get(op.cid); 119 - // if (!recordBytes) continue; 120 - // const record = cborToLexRecord(recordBytes); 121 - // const create = { uri, cid: op.cid.toString(), author: evt.repo }; 122 - // if (collection === ids.AppBskyFeedPost && isPost(record)) { 123 - // opsByType.posts.creates.push({ record, ...create }); 124 - // } else if (collection === ids.AppBskyFeedRepost && isRepost(record)) { 125 - // opsByType.reposts.creates.push({ record, ...create }); 126 - // } else if (collection === ids.AppBskyFeedLike && isLike(record)) { 127 - // opsByType.likes.creates.push({ record, ...create }); 128 - // } else if (collection === ids.AppBskyGraphFollow && isFollow(record)) { 129 - // opsByType.follows.creates.push({ record, ...create }); 130 - // } 131 - // } 132 - 133 - // if (op.action === "delete") { 134 - // if (collection === ids.AppBskyFeedPost) { 135 - // opsByType.posts.deletes.push({ uri }); 136 - // } else if (collection === ids.AppBskyFeedRepost) { 137 - // opsByType.reposts.deletes.push({ uri }); 138 - // } else if (collection === ids.AppBskyFeedLike) { 139 - // opsByType.likes.deletes.push({ uri }); 140 - // } else if (collection === ids.AppBskyGraphFollow) { 141 - // opsByType.follows.deletes.push({ uri }); 142 - // } 143 - // } 144 - // } 145 - 146 - // return opsByType; 147 - // }; 148 - 149 - // type OperationsByType = { 150 - // posts: Operations<PostRecord>; 151 - // reposts: Operations<RepostRecord>; 152 - // likes: Operations<LikeRecord>; 153 - // follows: Operations<FollowRecord>; 154 - // }; 155 - 156 - // type Operations<T = Record<string, unknown>> = { 157 - // creates: CreateOp<T>[]; 158 - // deletes: DeleteOp[]; 159 - // }; 160 - 161 - // type CreateOp<T> = { 162 - // uri: string; 163 - // cid: string; 164 - // author: string; 165 - // record: T; 166 - // }; 167 - 168 - // type DeleteOp = { 169 - // uri: string; 170 - // }; 171 - 172 - // export const isPost = (obj: unknown): obj is PostRecord => { 173 - // return isType(obj, ids.AppBskyFeedPost); 174 - // }; 175 - 176 - // export const isRepost = (obj: unknown): obj is RepostRecord => { 177 - // return isType(obj, ids.AppBskyFeedRepost); 178 - // }; 179 - 180 - // export const isLike = (obj: unknown): obj is LikeRecord => { 181 - // return isType(obj, ids.AppBskyFeedLike); 182 - // }; 183 - 184 - // export const isFollow = (obj: unknown): obj is FollowRecord => { 185 - // return isType(obj, ids.AppBskyGraphFollow); 186 - // }; 187 - 188 - // const isType = (obj: unknown, nsid: string) => { 189 - // try { 190 - // lexicons.assertValidRecord(nsid, fixBlobRefs(obj)); 191 - // return true; 192 - // } catch (err) { 193 - // return false; 194 - // } 195 - // }; 196 - 197 - // // @TODO right now record validation fails on BlobRefs 198 - // // simply because multiple packages have their own copy 199 - // // of the BlobRef class, causing instanceof checks to fail. 200 - // // This is a temporary solution. 201 - // const fixBlobRefs = (obj: unknown): unknown => { 202 - // if (Array.isArray(obj)) { 203 - // return obj.map(fixBlobRefs); 204 - // } 205 - // if (obj && typeof obj === "object") { 206 - // if (obj.constructor.name === "BlobRef") { 207 - // const blob = obj as BlobRef; 208 - // return new BlobRef(blob.ref, blob.mimeType, blob.size, blob.original); 209 - // } 210 - // return Object.entries(obj).reduce((acc, [key, val]) => { 211 - // return Object.assign(acc, { [key]: fixBlobRefs(val) }); 212 - // }, {} as Record<string, unknown>); 213 - // } 214 - // return obj; 215 - // };