danabra.mov
forked from
samuel.fm/statusphere-react
the statusphere demo reworked into a vite/react app in a monorepo
tweak linting
+6
biome.json
+6
biome.json
+7
-7
src/config.ts
+7
-7
src/config.ts
+11
-11
src/db/index.ts
+11
-11
src/db/index.ts
···
1
-
import SqliteDb from "better-sqlite3";
2
-
import { Kysely, Migrator, SqliteDialect } from "kysely";
3
-
import { migrationProvider } from "./migrations";
4
-
import type { DatabaseSchema } from "./schema";
5
6
export const createDb = (location: string): Database => {
7
return new Kysely<DatabaseSchema>({
8
dialect: new SqliteDialect({
9
database: new SqliteDb(location),
10
}),
11
-
});
12
-
};
13
14
export const migrateToLatest = async (db: Database) => {
15
-
const migrator = new Migrator({ db, provider: migrationProvider });
16
-
const { error } = await migrator.migrateToLatest();
17
-
if (error) throw error;
18
-
};
19
20
-
export type Database = Kysely<DatabaseSchema>;
···
1
+
import SqliteDb from 'better-sqlite3'
2
+
import { Kysely, Migrator, SqliteDialect } from 'kysely'
3
+
import { migrationProvider } from './migrations'
4
+
import type { DatabaseSchema } from './schema'
5
6
export const createDb = (location: string): Database => {
7
return new Kysely<DatabaseSchema>({
8
dialect: new SqliteDialect({
9
database: new SqliteDb(location),
10
}),
11
+
})
12
+
}
13
14
export const migrateToLatest = async (db: Database) => {
15
+
const migrator = new Migrator({ db, provider: migrationProvider })
16
+
const { error } = await migrator.migrateToLatest()
17
+
if (error) throw error
18
+
}
19
20
+
export type Database = Kysely<DatabaseSchema>
+12
-12
src/db/migrations.ts
+12
-12
src/db/migrations.ts
···
1
-
import type { Kysely, Migration, MigrationProvider } from "kysely";
2
3
-
const migrations: Record<string, Migration> = {};
4
5
export const migrationProvider: MigrationProvider = {
6
async getMigrations() {
7
-
return migrations;
8
},
9
-
};
10
11
-
migrations["001"] = {
12
async up(db: Kysely<unknown>) {
13
await db.schema
14
-
.createTable("post")
15
-
.addColumn("uri", "varchar", (col) => col.primaryKey())
16
-
.addColumn("text", "varchar", (col) => col.notNull())
17
-
.addColumn("indexedAt", "varchar", (col) => col.notNull())
18
-
.execute();
19
},
20
async down(db: Kysely<unknown>) {
21
-
await db.schema.dropTable("post").execute();
22
},
23
-
};
···
1
+
import type { Kysely, Migration, MigrationProvider } from 'kysely'
2
3
+
const migrations: Record<string, Migration> = {}
4
5
export const migrationProvider: MigrationProvider = {
6
async getMigrations() {
7
+
return migrations
8
},
9
+
}
10
11
+
migrations['001'] = {
12
async up(db: Kysely<unknown>) {
13
await db.schema
14
+
.createTable('post')
15
+
.addColumn('uri', 'varchar', (col) => col.primaryKey())
16
+
.addColumn('text', 'varchar', (col) => col.notNull())
17
+
.addColumn('indexedAt', 'varchar', (col) => col.notNull())
18
+
.execute()
19
},
20
async down(db: Kysely<unknown>) {
21
+
await db.schema.dropTable('post').execute()
22
},
23
+
}
+6
-6
src/db/schema.ts
+6
-6
src/db/schema.ts
+7
-7
src/env.ts
+7
-7
src/env.ts
···
1
-
import dotenv from "dotenv";
2
-
import { cleanEnv, host, num, port, str, testOnly } from "envalid";
3
4
-
dotenv.config();
5
6
export const env = cleanEnv(process.env, {
7
-
NODE_ENV: str({ devDefault: testOnly("test"), choices: ["development", "production", "test"] }),
8
-
HOST: host({ devDefault: testOnly("localhost") }),
9
PORT: port({ devDefault: testOnly(3000) }),
10
-
CORS_ORIGIN: str({ devDefault: testOnly("http://localhost:3000") }),
11
COMMON_RATE_LIMIT_MAX_REQUESTS: num({ devDefault: testOnly(1000) }),
12
COMMON_RATE_LIMIT_WINDOW_MS: num({ devDefault: testOnly(1000) }),
13
-
});
···
1
+
import dotenv from 'dotenv'
2
+
import { cleanEnv, host, num, port, str, testOnly } from 'envalid'
3
4
+
dotenv.config()
5
6
export const env = cleanEnv(process.env, {
7
+
NODE_ENV: str({ devDefault: testOnly('test'), choices: ['development', 'production', 'test'] }),
8
+
HOST: host({ devDefault: testOnly('localhost') }),
9
PORT: port({ devDefault: testOnly(3000) }),
10
+
CORS_ORIGIN: str({ devDefault: testOnly('http://localhost:3000') }),
11
COMMON_RATE_LIMIT_MAX_REQUESTS: num({ devDefault: testOnly(1000) }),
12
COMMON_RATE_LIMIT_WINDOW_MS: num({ devDefault: testOnly(1000) }),
13
+
})
+88
-94
src/firehose/firehose.ts
+88
-94
src/firehose/firehose.ts
···
1
-
import type { RepoRecord } from "@atproto/lexicon";
2
-
import { cborToLexRecord, readCar } from "@atproto/repo";
3
-
import { AtUri } from "@atproto/syntax";
4
-
import { Subscription } from "@atproto/xrpc-server";
5
-
import type { CID } from "multiformats/cid";
6
import {
7
type Account,
8
type Commit,
···
12
isCommit,
13
isIdentity,
14
isValidRepoEvent,
15
-
} from "./lexicons";
16
17
type Opts = {
18
-
service?: string;
19
-
getCursor?: () => Promise<number | undefined>;
20
-
setCursor?: (cursor: number) => Promise<void>;
21
-
subscriptionReconnectDelay?: number;
22
-
filterCollections?: string[];
23
-
excludeIdentity?: boolean;
24
-
excludeAccount?: boolean;
25
-
excludeCommit?: boolean;
26
-
};
27
28
export class Firehose {
29
-
public sub: Subscription<RepoEvent>;
30
-
private abortController: AbortController;
31
32
constructor(public opts: Opts) {
33
-
this.abortController = new AbortController();
34
this.sub = new Subscription({
35
-
service: opts.service ?? "https://bsky.network",
36
-
method: "com.atproto.sync.subscribeRepos",
37
signal: this.abortController.signal,
38
getParams: async () => {
39
-
if (!opts.getCursor) return undefined;
40
-
const cursor = await opts.getCursor();
41
-
return { cursor };
42
},
43
validate: (value: unknown) => {
44
try {
45
-
return isValidRepoEvent(value);
46
} catch (err) {
47
-
console.error("repo subscription skipped invalid message", err);
48
}
49
},
50
-
});
51
}
52
53
async *run(): AsyncGenerator<Event> {
···
55
for await (const evt of this.sub) {
56
try {
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
}
67
} else if (isAccount(evt) && !this.opts.excludeAccount) {
68
-
const parsed = parseAccount(evt);
69
if (parsed) {
70
-
yield parsed;
71
}
72
} else if (isIdentity(evt) && !this.opts.excludeIdentity) {
73
-
yield parseIdentity(evt);
74
}
75
} catch (err) {
76
-
console.error("repo subscription could not handle message", err);
77
}
78
-
if (this.opts.setCursor && typeof evt.seq === "number") {
79
-
await this.opts.setCursor(evt.seq);
80
}
81
}
82
} catch (err) {
83
-
console.error("repo subscription errored", err);
84
-
setTimeout(
85
-
() => this.run(),
86
-
this.opts.subscriptionReconnectDelay ?? 3000
87
-
);
88
}
89
}
90
91
destroy() {
92
-
this.abortController.abort();
93
}
94
}
95
96
export const parseCommit = async (evt: Commit): Promise<CommitEvt[]> => {
97
-
const car = await readCar(evt.blocks);
98
99
-
const evts: CommitEvt[] = [];
100
101
for (const op of evt.ops) {
102
-
const uri = new AtUri(`at://${evt.repo}/${op.path}`);
103
104
const meta: CommitMeta = {
105
uri,
106
author: uri.host,
107
collection: uri.collection,
108
rkey: uri.rkey,
109
-
};
110
111
-
if (op.action === "create" || op.action === "update") {
112
-
if (!op.cid) continue;
113
-
const recordBytes = car.blocks.get(op.cid);
114
-
if (!recordBytes) continue;
115
-
const record = cborToLexRecord(recordBytes);
116
evts.push({
117
...meta,
118
-
event: op.action as "create" | "update",
119
cid: op.cid,
120
record,
121
-
});
122
}
123
124
-
if (op.action === "delete") {
125
evts.push({
126
...meta,
127
-
event: "delete",
128
-
});
129
}
130
}
131
132
-
return evts;
133
-
};
134
135
export const parseIdentity = (evt: Identity): IdentityEvt => {
136
return {
137
-
event: "identity",
138
did: evt.did,
139
handle: evt.handle,
140
-
};
141
-
};
142
143
export const parseAccount = (evt: Account): AccountEvt | undefined => {
144
-
if (evt.status && !isValidStatus(evt.status)) return;
145
return {
146
-
event: "account",
147
did: evt.did,
148
active: evt.active,
149
status: evt.status as AccountStatus,
150
-
};
151
-
};
152
153
const isValidStatus = (str: string): str is AccountStatus => {
154
-
return ["takendown", "suspended", "deleted", "deactivated"].includes(str);
155
-
};
156
157
-
type Event = CommitEvt | IdentityEvt | AccountEvt;
158
159
type CommitMeta = {
160
-
uri: AtUri;
161
-
author: string;
162
-
collection: string;
163
-
rkey: string;
164
-
};
165
166
-
type CommitEvt = Create | Update | Delete;
167
168
type Create = CommitMeta & {
169
-
event: "create";
170
-
record: RepoRecord;
171
-
cid: CID;
172
-
};
173
174
type Update = CommitMeta & {
175
-
event: "update";
176
-
};
177
178
type Delete = CommitMeta & {
179
-
event: "delete";
180
-
};
181
182
type IdentityEvt = {
183
-
event: "identity";
184
-
did: string;
185
-
handle?: string;
186
-
};
187
188
type AccountEvt = {
189
-
event: "account";
190
-
did: string;
191
-
active: boolean;
192
-
status?: AccountStatus;
193
-
};
194
195
-
type AccountStatus = "takendown" | "suspended" | "deleted" | "deactivated";
···
1
+
import type { RepoRecord } from '@atproto/lexicon'
2
+
import { cborToLexRecord, readCar } from '@atproto/repo'
3
+
import { AtUri } from '@atproto/syntax'
4
+
import { Subscription } from '@atproto/xrpc-server'
5
+
import type { CID } from 'multiformats/cid'
6
import {
7
type Account,
8
type Commit,
···
12
isCommit,
13
isIdentity,
14
isValidRepoEvent,
15
+
} from './lexicons'
16
17
type Opts = {
18
+
service?: string
19
+
getCursor?: () => Promise<number | undefined>
20
+
setCursor?: (cursor: number) => Promise<void>
21
+
subscriptionReconnectDelay?: number
22
+
filterCollections?: string[]
23
+
excludeIdentity?: boolean
24
+
excludeAccount?: boolean
25
+
excludeCommit?: boolean
26
+
}
27
28
export class Firehose {
29
+
public sub: Subscription<RepoEvent>
30
+
private abortController: AbortController
31
32
constructor(public opts: Opts) {
33
+
this.abortController = new AbortController()
34
this.sub = new Subscription({
35
+
service: opts.service ?? 'https://bsky.network',
36
+
method: 'com.atproto.sync.subscribeRepos',
37
signal: this.abortController.signal,
38
getParams: async () => {
39
+
if (!opts.getCursor) return undefined
40
+
const cursor = await opts.getCursor()
41
+
return { cursor }
42
},
43
validate: (value: unknown) => {
44
try {
45
+
return isValidRepoEvent(value)
46
} catch (err) {
47
+
console.error('repo subscription skipped invalid message', err)
48
}
49
},
50
+
})
51
}
52
53
async *run(): AsyncGenerator<Event> {
···
55
for await (const evt of this.sub) {
56
try {
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
}
64
} else if (isAccount(evt) && !this.opts.excludeAccount) {
65
+
const parsed = parseAccount(evt)
66
if (parsed) {
67
+
yield parsed
68
}
69
} else if (isIdentity(evt) && !this.opts.excludeIdentity) {
70
+
yield parseIdentity(evt)
71
}
72
} catch (err) {
73
+
console.error('repo subscription could not handle message', err)
74
}
75
+
if (this.opts.setCursor && typeof evt.seq === 'number') {
76
+
await this.opts.setCursor(evt.seq)
77
}
78
}
79
} catch (err) {
80
+
console.error('repo subscription errored', err)
81
+
setTimeout(() => this.run(), this.opts.subscriptionReconnectDelay ?? 3000)
82
}
83
}
84
85
destroy() {
86
+
this.abortController.abort()
87
}
88
}
89
90
export const parseCommit = async (evt: Commit): Promise<CommitEvt[]> => {
91
+
const car = await readCar(evt.blocks)
92
93
+
const evts: CommitEvt[] = []
94
95
for (const op of evt.ops) {
96
+
const uri = new AtUri(`at://${evt.repo}/${op.path}`)
97
98
const meta: CommitMeta = {
99
uri,
100
author: uri.host,
101
collection: uri.collection,
102
rkey: uri.rkey,
103
+
}
104
105
+
if (op.action === 'create' || op.action === 'update') {
106
+
if (!op.cid) continue
107
+
const recordBytes = car.blocks.get(op.cid)
108
+
if (!recordBytes) continue
109
+
const record = cborToLexRecord(recordBytes)
110
evts.push({
111
...meta,
112
+
event: op.action as 'create' | 'update',
113
cid: op.cid,
114
record,
115
+
})
116
}
117
118
+
if (op.action === 'delete') {
119
evts.push({
120
...meta,
121
+
event: 'delete',
122
+
})
123
}
124
}
125
126
+
return evts
127
+
}
128
129
export const parseIdentity = (evt: Identity): IdentityEvt => {
130
return {
131
+
event: 'identity',
132
did: evt.did,
133
handle: evt.handle,
134
+
}
135
+
}
136
137
export const parseAccount = (evt: Account): AccountEvt | undefined => {
138
+
if (evt.status && !isValidStatus(evt.status)) return
139
return {
140
+
event: 'account',
141
did: evt.did,
142
active: evt.active,
143
status: evt.status as AccountStatus,
144
+
}
145
+
}
146
147
const isValidStatus = (str: string): str is AccountStatus => {
148
+
return ['takendown', 'suspended', 'deleted', 'deactivated'].includes(str)
149
+
}
150
151
+
type Event = CommitEvt | IdentityEvt | AccountEvt
152
153
type CommitMeta = {
154
+
uri: AtUri
155
+
author: string
156
+
collection: string
157
+
rkey: string
158
+
}
159
160
+
type CommitEvt = Create | Update | Delete
161
162
type Create = CommitMeta & {
163
+
event: 'create'
164
+
record: RepoRecord
165
+
cid: CID
166
+
}
167
168
type Update = CommitMeta & {
169
+
event: 'update'
170
+
}
171
172
type Delete = CommitMeta & {
173
+
event: 'delete'
174
+
}
175
176
type IdentityEvt = {
177
+
event: 'identity'
178
+
did: string
179
+
handle?: string
180
+
}
181
182
type AccountEvt = {
183
+
event: 'account'
184
+
did: string
185
+
active: boolean
186
+
status?: AccountStatus
187
+
}
188
189
+
type AccountStatus = 'takendown' | 'suspended' | 'deleted' | 'deactivated'
+10
-10
src/firehose/ingester.ts
+10
-10
src/firehose/ingester.ts
···
1
-
import type { Database } from "#/db";
2
-
import { Firehose } from "#/firehose/firehose";
3
4
export class Ingester {
5
-
firehose: Firehose | undefined;
6
constructor(public db: Database) {}
7
8
async start() {
9
-
const firehose = new Firehose({});
10
11
for await (const evt of firehose.run()) {
12
-
if (evt.event === "create") {
13
-
if (evt.collection !== "app.bsky.feed.post") continue;
14
-
const post: any = evt.record; // @TODO fix types
15
await this.db
16
-
.insertInto("post")
17
.values({
18
uri: evt.uri.toString(),
19
text: post.text as string,
20
indexedAt: new Date().toISOString(),
21
})
22
-
.execute();
23
}
24
}
25
}
26
27
destroy() {
28
-
this.firehose?.destroy();
29
}
30
}
···
1
+
import type { Database } from '#/db'
2
+
import { Firehose } from '#/firehose/firehose'
3
4
export class Ingester {
5
+
firehose: Firehose | undefined
6
constructor(public db: Database) {}
7
8
async start() {
9
+
const firehose = new Firehose({})
10
11
for await (const evt of firehose.run()) {
12
+
if (evt.event === 'create') {
13
+
if (evt.collection !== 'app.bsky.feed.post') continue
14
+
const post: any = evt.record // @TODO fix types
15
await this.db
16
+
.insertInto('post')
17
.values({
18
uri: evt.uri.toString(),
19
text: post.text as string,
20
indexedAt: new Date().toISOString(),
21
})
22
+
.execute()
23
}
24
}
25
}
26
27
destroy() {
28
+
this.firehose?.destroy()
29
}
30
}
+148
-205
src/firehose/lexicons.ts
+148
-205
src/firehose/lexicons.ts
···
1
-
import type { IncomingMessage } from "node:http";
2
3
-
import { type LexiconDoc, Lexicons } from "@atproto/lexicon";
4
-
import type { ErrorFrame, HandlerAuth } from "@atproto/xrpc-server";
5
-
import type { CID } from "multiformats/cid";
6
7
// @NOTE: this file is an ugly copy job of codegen output. I'd like to clean this whole thing up
8
9
export function isObj(v: unknown): v is Record<string, unknown> {
10
-
return typeof v === "object" && v !== null;
11
}
12
13
-
export function hasProp<K extends PropertyKey>(
14
-
data: object,
15
-
prop: K
16
-
): data is Record<K, unknown> {
17
-
return prop in data;
18
}
19
20
export interface QueryParams {
21
/** The last known event seq number to backfill from. */
22
-
cursor?: number;
23
}
24
25
export type RepoEvent =
···
30
| Migrate
31
| Tombstone
32
| Info
33
-
| { $type: string; [k: string]: unknown };
34
-
export type HandlerError = ErrorFrame<"FutureCursor" | "ConsumerTooSlow">;
35
-
export type HandlerOutput = HandlerError | RepoEvent;
36
export type HandlerReqCtx<HA extends HandlerAuth = never> = {
37
-
auth: HA;
38
-
params: QueryParams;
39
-
req: IncomingMessage;
40
-
signal: AbortSignal;
41
-
};
42
-
export type Handler<HA extends HandlerAuth = never> = (
43
-
ctx: HandlerReqCtx<HA>
44
-
) => AsyncIterable<HandlerOutput>;
45
46
/** Represents an update of repository state. Note that empty commits are allowed, which include no repo data changes, but an update to rev and signature. */
47
export interface Commit {
48
/** The stream sequence number of this message. */
49
-
seq: number;
50
/** DEPRECATED -- unused */
51
-
rebase: boolean;
52
/** Indicates that this commit contained too many ops, or data size was too large. Consumers will need to make a separate request to get missing data. */
53
-
tooBig: boolean;
54
/** The repo this event comes from. */
55
-
repo: string;
56
/** Repo commit object CID. */
57
-
commit: CID;
58
/** DEPRECATED -- unused. WARNING -- nullable and optional; stick with optional to ensure golang interoperability. */
59
-
prev?: CID | null;
60
/** The rev of the emitted commit. Note that this information is also in the commit object included in blocks, unless this is a tooBig event. */
61
-
rev: string;
62
/** The rev of the last emitted commit from this repo (if any). */
63
-
since: string | null;
64
/** CAR file containing relevant blocks, as a diff since the previous repo state. */
65
-
blocks: Uint8Array;
66
-
ops: RepoOp[];
67
-
blobs: CID[];
68
/** Timestamp of when this message was originally broadcast. */
69
-
time: string;
70
-
[k: string]: unknown;
71
}
72
73
export function isCommit(v: unknown): v is Commit {
74
-
return (
75
-
isObj(v) &&
76
-
hasProp(v, "$type") &&
77
-
v.$type === "com.atproto.sync.subscribeRepos#commit"
78
-
);
79
}
80
81
/** Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache. */
82
export interface Identity {
83
-
seq: number;
84
-
did: string;
85
-
time: string;
86
/** The current handle for the account, or 'handle.invalid' if validation fails. This field is optional, might have been validated or passed-through from an upstream source. Semantics and behaviors for PDS vs Relay may evolve in the future; see atproto specs for more details. */
87
-
handle?: string;
88
-
[k: string]: unknown;
89
}
90
91
export function isIdentity(v: unknown): v is Identity {
92
-
return (
93
-
isObj(v) &&
94
-
hasProp(v, "$type") &&
95
-
v.$type === "com.atproto.sync.subscribeRepos#identity"
96
-
);
97
}
98
99
/** Represents a change to an account's status on a host (eg, PDS or Relay). The semantics of this event are that the status is at the host which emitted the event, not necessarily that at the currently active PDS. Eg, a Relay takedown would emit a takedown with active=false, even if the PDS is still active. */
100
export interface Account {
101
-
seq: number;
102
-
did: string;
103
-
time: string;
104
/** Indicates that the account has a repository which can be fetched from the host that emitted this event. */
105
-
active: boolean;
106
/** If active=false, this optional field indicates a reason for why the account is not active. */
107
-
status?:
108
-
| "takendown"
109
-
| "suspended"
110
-
| "deleted"
111
-
| "deactivated"
112
-
| (string & {});
113
-
[k: string]: unknown;
114
}
115
116
export function isAccount(v: unknown): v is Account {
117
-
return (
118
-
isObj(v) &&
119
-
hasProp(v, "$type") &&
120
-
v.$type === "com.atproto.sync.subscribeRepos#account"
121
-
);
122
}
123
124
/** DEPRECATED -- Use #identity event instead */
125
export interface Handle {
126
-
seq: number;
127
-
did: string;
128
-
handle: string;
129
-
time: string;
130
-
[k: string]: unknown;
131
}
132
133
export function isHandle(v: unknown): v is Handle {
134
-
return (
135
-
isObj(v) &&
136
-
hasProp(v, "$type") &&
137
-
v.$type === "com.atproto.sync.subscribeRepos#handle"
138
-
);
139
}
140
141
/** DEPRECATED -- Use #account event instead */
142
export interface Migrate {
143
-
seq: number;
144
-
did: string;
145
-
migrateTo: string | null;
146
-
time: string;
147
-
[k: string]: unknown;
148
}
149
150
export function isMigrate(v: unknown): v is Migrate {
151
-
return (
152
-
isObj(v) &&
153
-
hasProp(v, "$type") &&
154
-
v.$type === "com.atproto.sync.subscribeRepos#migrate"
155
-
);
156
}
157
158
/** DEPRECATED -- Use #account event instead */
159
export interface Tombstone {
160
-
seq: number;
161
-
did: string;
162
-
time: string;
163
-
[k: string]: unknown;
164
}
165
166
export function isTombstone(v: unknown): v is Tombstone {
167
-
return (
168
-
isObj(v) &&
169
-
hasProp(v, "$type") &&
170
-
v.$type === "com.atproto.sync.subscribeRepos#tombstone"
171
-
);
172
}
173
174
export interface Info {
175
-
name: "OutdatedCursor" | (string & {});
176
-
message?: string;
177
-
[k: string]: unknown;
178
}
179
180
export function isInfo(v: unknown): v is Info {
181
-
return (
182
-
isObj(v) &&
183
-
hasProp(v, "$type") &&
184
-
v.$type === "com.atproto.sync.subscribeRepos#info"
185
-
);
186
}
187
188
/** A repo operation, ie a mutation of a single record. */
189
export interface RepoOp {
190
-
action: "create" | "update" | "delete" | (string & {});
191
-
path: string;
192
/** For creates and updates, the new record CID. For deletions, null. */
193
-
cid: CID | null;
194
-
[k: string]: unknown;
195
}
196
197
export function isRepoOp(v: unknown): v is RepoOp {
198
-
return (
199
-
isObj(v) &&
200
-
hasProp(v, "$type") &&
201
-
v.$type === "com.atproto.sync.subscribeRepos#repoOp"
202
-
);
203
}
204
205
export const ComAtprotoSyncSubscribeRepos: LexiconDoc = {
206
lexicon: 1,
207
-
id: "com.atproto.sync.subscribeRepos",
208
defs: {
209
main: {
210
-
type: "subscription",
211
-
description: "Subscribe to repo updates",
212
parameters: {
213
-
type: "params",
214
properties: {
215
cursor: {
216
-
type: "integer",
217
-
description: "The last known event to backfill from.",
218
},
219
},
220
},
221
message: {
222
schema: {
223
-
type: "union",
224
refs: [
225
-
"lex:com.atproto.sync.subscribeRepos#commit",
226
-
"lex:com.atproto.sync.subscribeRepos#handle",
227
-
"lex:com.atproto.sync.subscribeRepos#migrate",
228
-
"lex:com.atproto.sync.subscribeRepos#tombstone",
229
-
"lex:com.atproto.sync.subscribeRepos#info",
230
],
231
},
232
},
233
errors: [
234
{
235
-
name: "FutureCursor",
236
},
237
{
238
-
name: "ConsumerTooSlow",
239
},
240
],
241
},
242
commit: {
243
-
type: "object",
244
-
required: [
245
-
"seq",
246
-
"rebase",
247
-
"tooBig",
248
-
"repo",
249
-
"commit",
250
-
"rev",
251
-
"since",
252
-
"blocks",
253
-
"ops",
254
-
"blobs",
255
-
"time",
256
-
],
257
-
nullable: ["prev", "since"],
258
properties: {
259
seq: {
260
-
type: "integer",
261
},
262
rebase: {
263
-
type: "boolean",
264
},
265
tooBig: {
266
-
type: "boolean",
267
},
268
repo: {
269
-
type: "string",
270
-
format: "did",
271
},
272
commit: {
273
-
type: "cid-link",
274
},
275
prev: {
276
-
type: "cid-link",
277
},
278
rev: {
279
-
type: "string",
280
-
description: "The rev of the emitted commit",
281
},
282
since: {
283
-
type: "string",
284
-
description: "The rev of the last emitted commit from this repo",
285
},
286
blocks: {
287
-
type: "bytes",
288
-
description: "CAR file containing relevant blocks",
289
maxLength: 1000000,
290
},
291
ops: {
292
-
type: "array",
293
items: {
294
-
type: "ref",
295
-
ref: "lex:com.atproto.sync.subscribeRepos#repoOp",
296
},
297
maxLength: 200,
298
},
299
blobs: {
300
-
type: "array",
301
items: {
302
-
type: "cid-link",
303
},
304
},
305
time: {
306
-
type: "string",
307
-
format: "datetime",
308
},
309
},
310
},
311
handle: {
312
-
type: "object",
313
-
required: ["seq", "did", "handle", "time"],
314
properties: {
315
seq: {
316
-
type: "integer",
317
},
318
did: {
319
-
type: "string",
320
-
format: "did",
321
},
322
handle: {
323
-
type: "string",
324
-
format: "handle",
325
},
326
time: {
327
-
type: "string",
328
-
format: "datetime",
329
},
330
},
331
},
332
migrate: {
333
-
type: "object",
334
-
required: ["seq", "did", "migrateTo", "time"],
335
-
nullable: ["migrateTo"],
336
properties: {
337
seq: {
338
-
type: "integer",
339
},
340
did: {
341
-
type: "string",
342
-
format: "did",
343
},
344
migrateTo: {
345
-
type: "string",
346
},
347
time: {
348
-
type: "string",
349
-
format: "datetime",
350
},
351
},
352
},
353
tombstone: {
354
-
type: "object",
355
-
required: ["seq", "did", "time"],
356
properties: {
357
seq: {
358
-
type: "integer",
359
},
360
did: {
361
-
type: "string",
362
-
format: "did",
363
},
364
time: {
365
-
type: "string",
366
-
format: "datetime",
367
},
368
},
369
},
370
info: {
371
-
type: "object",
372
-
required: ["name"],
373
properties: {
374
name: {
375
-
type: "string",
376
-
knownValues: ["OutdatedCursor"],
377
},
378
message: {
379
-
type: "string",
380
},
381
},
382
},
383
repoOp: {
384
-
type: "object",
385
description:
386
"A repo operation, ie a write of a single record. For creates and updates, cid is the record's CID as of this operation. For deletes, it's null.",
387
-
required: ["action", "path", "cid"],
388
-
nullable: ["cid"],
389
properties: {
390
action: {
391
-
type: "string",
392
-
knownValues: ["create", "update", "delete"],
393
},
394
path: {
395
-
type: "string",
396
},
397
cid: {
398
-
type: "cid-link",
399
},
400
},
401
},
402
},
403
-
};
404
405
-
const lexicons = new Lexicons([ComAtprotoSyncSubscribeRepos]);
406
407
export const isValidRepoEvent = (evt: unknown) => {
408
-
return lexicons.assertValidXrpcMessage<RepoEvent>(
409
-
"com.atproto.sync.subscribeRepos",
410
-
evt
411
-
);
412
-
};
···
1
+
import type { IncomingMessage } from 'node:http'
2
3
+
import { type LexiconDoc, Lexicons } from '@atproto/lexicon'
4
+
import type { ErrorFrame, HandlerAuth } from '@atproto/xrpc-server'
5
+
import type { CID } from 'multiformats/cid'
6
7
// @NOTE: this file is an ugly copy job of codegen output. I'd like to clean this whole thing up
8
9
export function isObj(v: unknown): v is Record<string, unknown> {
10
+
return typeof v === 'object' && v !== null
11
}
12
13
+
export function hasProp<K extends PropertyKey>(data: object, prop: K): data is Record<K, unknown> {
14
+
return prop in data
15
}
16
17
export interface QueryParams {
18
/** The last known event seq number to backfill from. */
19
+
cursor?: number
20
}
21
22
export type RepoEvent =
···
27
| Migrate
28
| Tombstone
29
| Info
30
+
| { $type: string; [k: string]: unknown }
31
+
export type HandlerError = ErrorFrame<'FutureCursor' | 'ConsumerTooSlow'>
32
+
export type HandlerOutput = HandlerError | RepoEvent
33
export type HandlerReqCtx<HA extends HandlerAuth = never> = {
34
+
auth: HA
35
+
params: QueryParams
36
+
req: IncomingMessage
37
+
signal: AbortSignal
38
+
}
39
+
export type Handler<HA extends HandlerAuth = never> = (ctx: HandlerReqCtx<HA>) => AsyncIterable<HandlerOutput>
40
41
/** Represents an update of repository state. Note that empty commits are allowed, which include no repo data changes, but an update to rev and signature. */
42
export interface Commit {
43
/** The stream sequence number of this message. */
44
+
seq: number
45
/** DEPRECATED -- unused */
46
+
rebase: boolean
47
/** Indicates that this commit contained too many ops, or data size was too large. Consumers will need to make a separate request to get missing data. */
48
+
tooBig: boolean
49
/** The repo this event comes from. */
50
+
repo: string
51
/** Repo commit object CID. */
52
+
commit: CID
53
/** DEPRECATED -- unused. WARNING -- nullable and optional; stick with optional to ensure golang interoperability. */
54
+
prev?: CID | null
55
/** The rev of the emitted commit. Note that this information is also in the commit object included in blocks, unless this is a tooBig event. */
56
+
rev: string
57
/** The rev of the last emitted commit from this repo (if any). */
58
+
since: string | null
59
/** CAR file containing relevant blocks, as a diff since the previous repo state. */
60
+
blocks: Uint8Array
61
+
ops: RepoOp[]
62
+
blobs: CID[]
63
/** Timestamp of when this message was originally broadcast. */
64
+
time: string
65
+
[k: string]: unknown
66
}
67
68
export function isCommit(v: unknown): v is Commit {
69
+
return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#commit'
70
}
71
72
/** Represents a change to an account's identity. Could be an updated handle, signing key, or pds hosting endpoint. Serves as a prod to all downstream services to refresh their identity cache. */
73
export interface Identity {
74
+
seq: number
75
+
did: string
76
+
time: string
77
/** The current handle for the account, or 'handle.invalid' if validation fails. This field is optional, might have been validated or passed-through from an upstream source. Semantics and behaviors for PDS vs Relay may evolve in the future; see atproto specs for more details. */
78
+
handle?: string
79
+
[k: string]: unknown
80
}
81
82
export function isIdentity(v: unknown): v is Identity {
83
+
return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#identity'
84
}
85
86
/** Represents a change to an account's status on a host (eg, PDS or Relay). The semantics of this event are that the status is at the host which emitted the event, not necessarily that at the currently active PDS. Eg, a Relay takedown would emit a takedown with active=false, even if the PDS is still active. */
87
export interface Account {
88
+
seq: number
89
+
did: string
90
+
time: string
91
/** Indicates that the account has a repository which can be fetched from the host that emitted this event. */
92
+
active: boolean
93
/** If active=false, this optional field indicates a reason for why the account is not active. */
94
+
status?: 'takendown' | 'suspended' | 'deleted' | 'deactivated' | (string & {})
95
+
[k: string]: unknown
96
}
97
98
export function isAccount(v: unknown): v is Account {
99
+
return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#account'
100
}
101
102
/** DEPRECATED -- Use #identity event instead */
103
export interface Handle {
104
+
seq: number
105
+
did: string
106
+
handle: string
107
+
time: string
108
+
[k: string]: unknown
109
}
110
111
export function isHandle(v: unknown): v is Handle {
112
+
return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#handle'
113
}
114
115
/** DEPRECATED -- Use #account event instead */
116
export interface Migrate {
117
+
seq: number
118
+
did: string
119
+
migrateTo: string | null
120
+
time: string
121
+
[k: string]: unknown
122
}
123
124
export function isMigrate(v: unknown): v is Migrate {
125
+
return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#migrate'
126
}
127
128
/** DEPRECATED -- Use #account event instead */
129
export interface Tombstone {
130
+
seq: number
131
+
did: string
132
+
time: string
133
+
[k: string]: unknown
134
}
135
136
export function isTombstone(v: unknown): v is Tombstone {
137
+
return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#tombstone'
138
}
139
140
export interface Info {
141
+
name: 'OutdatedCursor' | (string & {})
142
+
message?: string
143
+
[k: string]: unknown
144
}
145
146
export function isInfo(v: unknown): v is Info {
147
+
return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#info'
148
}
149
150
/** A repo operation, ie a mutation of a single record. */
151
export interface RepoOp {
152
+
action: 'create' | 'update' | 'delete' | (string & {})
153
+
path: string
154
/** For creates and updates, the new record CID. For deletions, null. */
155
+
cid: CID | null
156
+
[k: string]: unknown
157
}
158
159
export function isRepoOp(v: unknown): v is RepoOp {
160
+
return isObj(v) && hasProp(v, '$type') && v.$type === 'com.atproto.sync.subscribeRepos#repoOp'
161
}
162
163
export const ComAtprotoSyncSubscribeRepos: LexiconDoc = {
164
lexicon: 1,
165
+
id: 'com.atproto.sync.subscribeRepos',
166
defs: {
167
main: {
168
+
type: 'subscription',
169
+
description: 'Subscribe to repo updates',
170
parameters: {
171
+
type: 'params',
172
properties: {
173
cursor: {
174
+
type: 'integer',
175
+
description: 'The last known event to backfill from.',
176
},
177
},
178
},
179
message: {
180
schema: {
181
+
type: 'union',
182
refs: [
183
+
'lex:com.atproto.sync.subscribeRepos#commit',
184
+
'lex:com.atproto.sync.subscribeRepos#handle',
185
+
'lex:com.atproto.sync.subscribeRepos#migrate',
186
+
'lex:com.atproto.sync.subscribeRepos#tombstone',
187
+
'lex:com.atproto.sync.subscribeRepos#info',
188
],
189
},
190
},
191
errors: [
192
{
193
+
name: 'FutureCursor',
194
},
195
{
196
+
name: 'ConsumerTooSlow',
197
},
198
],
199
},
200
commit: {
201
+
type: 'object',
202
+
required: ['seq', 'rebase', 'tooBig', 'repo', 'commit', 'rev', 'since', 'blocks', 'ops', 'blobs', 'time'],
203
+
nullable: ['prev', 'since'],
204
properties: {
205
seq: {
206
+
type: 'integer',
207
},
208
rebase: {
209
+
type: 'boolean',
210
},
211
tooBig: {
212
+
type: 'boolean',
213
},
214
repo: {
215
+
type: 'string',
216
+
format: 'did',
217
},
218
commit: {
219
+
type: 'cid-link',
220
},
221
prev: {
222
+
type: 'cid-link',
223
},
224
rev: {
225
+
type: 'string',
226
+
description: 'The rev of the emitted commit',
227
},
228
since: {
229
+
type: 'string',
230
+
description: 'The rev of the last emitted commit from this repo',
231
},
232
blocks: {
233
+
type: 'bytes',
234
+
description: 'CAR file containing relevant blocks',
235
maxLength: 1000000,
236
},
237
ops: {
238
+
type: 'array',
239
items: {
240
+
type: 'ref',
241
+
ref: 'lex:com.atproto.sync.subscribeRepos#repoOp',
242
},
243
maxLength: 200,
244
},
245
blobs: {
246
+
type: 'array',
247
items: {
248
+
type: 'cid-link',
249
},
250
},
251
time: {
252
+
type: 'string',
253
+
format: 'datetime',
254
},
255
},
256
},
257
handle: {
258
+
type: 'object',
259
+
required: ['seq', 'did', 'handle', 'time'],
260
properties: {
261
seq: {
262
+
type: 'integer',
263
},
264
did: {
265
+
type: 'string',
266
+
format: 'did',
267
},
268
handle: {
269
+
type: 'string',
270
+
format: 'handle',
271
},
272
time: {
273
+
type: 'string',
274
+
format: 'datetime',
275
},
276
},
277
},
278
migrate: {
279
+
type: 'object',
280
+
required: ['seq', 'did', 'migrateTo', 'time'],
281
+
nullable: ['migrateTo'],
282
properties: {
283
seq: {
284
+
type: 'integer',
285
},
286
did: {
287
+
type: 'string',
288
+
format: 'did',
289
},
290
migrateTo: {
291
+
type: 'string',
292
},
293
time: {
294
+
type: 'string',
295
+
format: 'datetime',
296
},
297
},
298
},
299
tombstone: {
300
+
type: 'object',
301
+
required: ['seq', 'did', 'time'],
302
properties: {
303
seq: {
304
+
type: 'integer',
305
},
306
did: {
307
+
type: 'string',
308
+
format: 'did',
309
},
310
time: {
311
+
type: 'string',
312
+
format: 'datetime',
313
},
314
},
315
},
316
info: {
317
+
type: 'object',
318
+
required: ['name'],
319
properties: {
320
name: {
321
+
type: 'string',
322
+
knownValues: ['OutdatedCursor'],
323
},
324
message: {
325
+
type: 'string',
326
},
327
},
328
},
329
repoOp: {
330
+
type: 'object',
331
description:
332
"A repo operation, ie a write of a single record. For creates and updates, cid is the record's CID as of this operation. For deletes, it's null.",
333
+
required: ['action', 'path', 'cid'],
334
+
nullable: ['cid'],
335
properties: {
336
action: {
337
+
type: 'string',
338
+
knownValues: ['create', 'update', 'delete'],
339
},
340
path: {
341
+
type: 'string',
342
},
343
cid: {
344
+
type: 'cid-link',
345
},
346
},
347
},
348
},
349
+
}
350
351
+
const lexicons = new Lexicons([ComAtprotoSyncSubscribeRepos])
352
353
export const isValidRepoEvent = (evt: unknown) => {
354
+
return lexicons.assertValidXrpcMessage<RepoEvent>('com.atproto.sync.subscribeRepos', evt)
355
+
}
+10
-10
src/index.ts
+10
-10
src/index.ts
···
1
-
import { Server } from "#/server";
2
3
const run = async () => {
4
-
const server = await Server.create();
5
6
const onCloseSignal = async () => {
7
-
setTimeout(() => process.exit(1), 10000).unref(); // Force shutdown after 10s
8
-
await server.close();
9
-
process.exit();
10
-
};
11
12
-
process.on("SIGINT", onCloseSignal);
13
-
process.on("SIGTERM", onCloseSignal);
14
-
};
15
16
-
run();
···
1
+
import { Server } from '#/server'
2
3
const run = async () => {
4
+
const server = await Server.create()
5
6
const onCloseSignal = async () => {
7
+
setTimeout(() => process.exit(1), 10000).unref() // Force shutdown after 10s
8
+
await server.close()
9
+
process.exit()
10
+
}
11
12
+
process.on('SIGINT', onCloseSignal)
13
+
process.on('SIGTERM', onCloseSignal)
14
+
}
15
16
+
run()
+8
-8
src/middleware/errorHandler.ts
+8
-8
src/middleware/errorHandler.ts
···
1
-
import type { ErrorRequestHandler, RequestHandler } from "express";
2
-
import { StatusCodes } from "http-status-codes";
3
4
const unexpectedRequest: RequestHandler = (_req, res) => {
5
-
res.sendStatus(StatusCodes.NOT_FOUND);
6
-
};
7
8
const addErrorToRequestLog: ErrorRequestHandler = (err, _req, res, next) => {
9
-
res.locals.err = err;
10
-
next(err);
11
-
};
12
13
-
export default () => [unexpectedRequest, addErrorToRequestLog];
···
1
+
import type { ErrorRequestHandler, RequestHandler } from 'express'
2
+
import { StatusCodes } from 'http-status-codes'
3
4
const unexpectedRequest: RequestHandler = (_req, res) => {
5
+
res.sendStatus(StatusCodes.NOT_FOUND)
6
+
}
7
8
const addErrorToRequestLog: ErrorRequestHandler = (err, _req, res, next) => {
9
+
res.locals.err = err
10
+
next(err)
11
+
}
12
13
+
export default () => [unexpectedRequest, addErrorToRequestLog]
+52
-52
src/middleware/requestLogger.ts
+52
-52
src/middleware/requestLogger.ts
···
1
-
import { randomUUID } from "node:crypto";
2
-
import type { IncomingMessage, ServerResponse } from "node:http";
3
-
import type { Request, RequestHandler, Response } from "express";
4
-
import { StatusCodes, getReasonPhrase } from "http-status-codes";
5
-
import type { LevelWithSilent } from "pino";
6
-
import { type CustomAttributeKeys, type Options, pinoHttp } from "pino-http";
7
8
-
import { env } from "#/common/utils/envConfig";
9
10
enum LogLevel {
11
-
Fatal = "fatal",
12
-
Error = "error",
13
-
Warn = "warn",
14
-
Info = "info",
15
-
Debug = "debug",
16
-
Trace = "trace",
17
-
Silent = "silent",
18
}
19
20
type PinoCustomProps = {
21
-
request: Request;
22
-
response: Response;
23
-
error: Error;
24
-
responseBody: unknown;
25
-
};
26
27
const requestLogger = (options?: Options): RequestHandler[] => {
28
const pinoOptions: Options = {
29
enabled: env.isProduction,
30
-
customProps: customProps as unknown as Options["customProps"],
31
redact: [],
32
genReqId,
33
customLogLevel,
···
36
customErrorMessage: (_req, res) => `request errored with status code: ${res.statusCode}`,
37
customAttributeKeys,
38
...options,
39
-
};
40
-
return [responseBodyMiddleware, pinoHttp(pinoOptions)];
41
-
};
42
43
const customAttributeKeys: CustomAttributeKeys = {
44
-
req: "request",
45
-
res: "response",
46
-
err: "error",
47
-
responseTime: "timeTaken",
48
-
};
49
50
const customProps = (req: Request, res: Response): PinoCustomProps => ({
51
request: req,
52
response: res,
53
error: res.locals.err,
54
responseBody: res.locals.responseBody,
55
-
});
56
57
const responseBodyMiddleware: RequestHandler = (_req, res, next) => {
58
-
const isNotProduction = !env.isProduction;
59
if (isNotProduction) {
60
-
const originalSend = res.send;
61
res.send = (content) => {
62
-
res.locals.responseBody = content;
63
-
res.send = originalSend;
64
-
return originalSend.call(res, content);
65
-
};
66
}
67
-
next();
68
-
};
69
70
const customLogLevel = (_req: IncomingMessage, res: ServerResponse<IncomingMessage>, err?: Error): LevelWithSilent => {
71
-
if (err || res.statusCode >= StatusCodes.INTERNAL_SERVER_ERROR) return LogLevel.Error;
72
-
if (res.statusCode >= StatusCodes.BAD_REQUEST) return LogLevel.Warn;
73
-
if (res.statusCode >= StatusCodes.MULTIPLE_CHOICES) return LogLevel.Silent;
74
-
return LogLevel.Info;
75
-
};
76
77
const customSuccessMessage = (req: IncomingMessage, res: ServerResponse<IncomingMessage>) => {
78
-
if (res.statusCode === StatusCodes.NOT_FOUND) return getReasonPhrase(StatusCodes.NOT_FOUND);
79
-
return `${req.method} completed`;
80
-
};
81
82
const genReqId = (req: IncomingMessage, res: ServerResponse<IncomingMessage>) => {
83
-
const existingID = req.id ?? req.headers["x-request-id"];
84
-
if (existingID) return existingID;
85
-
const id = randomUUID();
86
-
res.setHeader("X-Request-Id", id);
87
-
return id;
88
-
};
89
90
-
export default requestLogger();
···
1
+
import { randomUUID } from 'node:crypto'
2
+
import type { IncomingMessage, ServerResponse } from 'node:http'
3
+
import type { Request, RequestHandler, Response } from 'express'
4
+
import { StatusCodes, getReasonPhrase } from 'http-status-codes'
5
+
import type { LevelWithSilent } from 'pino'
6
+
import { type CustomAttributeKeys, type Options, pinoHttp } from 'pino-http'
7
8
+
import { env } from '#/env'
9
10
enum LogLevel {
11
+
Fatal = 'fatal',
12
+
Error = 'error',
13
+
Warn = 'warn',
14
+
Info = 'info',
15
+
Debug = 'debug',
16
+
Trace = 'trace',
17
+
Silent = 'silent',
18
}
19
20
type PinoCustomProps = {
21
+
request: Request
22
+
response: Response
23
+
error: Error
24
+
responseBody: unknown
25
+
}
26
27
const requestLogger = (options?: Options): RequestHandler[] => {
28
const pinoOptions: Options = {
29
enabled: env.isProduction,
30
+
customProps: customProps as unknown as Options['customProps'],
31
redact: [],
32
genReqId,
33
customLogLevel,
···
36
customErrorMessage: (_req, res) => `request errored with status code: ${res.statusCode}`,
37
customAttributeKeys,
38
...options,
39
+
}
40
+
return [responseBodyMiddleware, pinoHttp(pinoOptions)]
41
+
}
42
43
const customAttributeKeys: CustomAttributeKeys = {
44
+
req: 'request',
45
+
res: 'response',
46
+
err: 'error',
47
+
responseTime: 'timeTaken',
48
+
}
49
50
const customProps = (req: Request, res: Response): PinoCustomProps => ({
51
request: req,
52
response: res,
53
error: res.locals.err,
54
responseBody: res.locals.responseBody,
55
+
})
56
57
const responseBodyMiddleware: RequestHandler = (_req, res, next) => {
58
+
const isNotProduction = !env.isProduction
59
if (isNotProduction) {
60
+
const originalSend = res.send
61
res.send = (content) => {
62
+
res.locals.responseBody = content
63
+
res.send = originalSend
64
+
return originalSend.call(res, content)
65
+
}
66
}
67
+
next()
68
+
}
69
70
const customLogLevel = (_req: IncomingMessage, res: ServerResponse<IncomingMessage>, err?: Error): LevelWithSilent => {
71
+
if (err || res.statusCode >= StatusCodes.INTERNAL_SERVER_ERROR) return LogLevel.Error
72
+
if (res.statusCode >= StatusCodes.BAD_REQUEST) return LogLevel.Warn
73
+
if (res.statusCode >= StatusCodes.MULTIPLE_CHOICES) return LogLevel.Silent
74
+
return LogLevel.Info
75
+
}
76
77
const customSuccessMessage = (req: IncomingMessage, res: ServerResponse<IncomingMessage>) => {
78
+
if (res.statusCode === StatusCodes.NOT_FOUND) return getReasonPhrase(StatusCodes.NOT_FOUND)
79
+
return `${req.method} completed`
80
+
}
81
82
const genReqId = (req: IncomingMessage, res: ServerResponse<IncomingMessage>) => {
83
+
const existingID = req.id ?? req.headers['x-request-id']
84
+
if (existingID) return existingID
85
+
const id = randomUUID()
86
+
res.setHeader('X-Request-Id', id)
87
+
return id
88
+
}
89
90
+
export default requestLogger()
+12
-17
src/routes/index.ts
+12
-17
src/routes/index.ts
···
1
-
import express from "express";
2
-
import type { AppContext } from "#/config";
3
-
import { handler } from "./util";
4
5
export const createRouter = (ctx: AppContext) => {
6
-
const router = express.Router();
7
8
router.get(
9
-
"/",
10
handler(async (req, res) => {
11
-
const posts = await ctx.db
12
-
.selectFrom("post")
13
-
.selectAll()
14
-
.orderBy("indexedAt", "desc")
15
-
.limit(10)
16
-
.execute();
17
-
const postTexts = posts.map((row) => row.text);
18
-
res.json(postTexts);
19
-
})
20
-
);
21
22
-
return router;
23
-
};
···
1
+
import express from 'express'
2
+
import type { AppContext } from '#/config'
3
+
import { handler } from './util'
4
5
export const createRouter = (ctx: AppContext) => {
6
+
const router = express.Router()
7
8
router.get(
9
+
'/',
10
handler(async (req, res) => {
11
+
const posts = await ctx.db.selectFrom('post').selectAll().orderBy('indexedAt', 'desc').limit(10).execute()
12
+
const postTexts = posts.map((row) => row.text)
13
+
res.json(postTexts)
14
+
}),
15
+
)
16
17
+
return router
18
+
}
+5
-10
src/routes/util.ts
+5
-10
src/routes/util.ts
+41
-41
src/server.ts
+41
-41
src/server.ts
···
1
-
import events from "node:events";
2
-
import type http from "node:http";
3
-
import cors from "cors";
4
-
import express, { type Express } from "express";
5
-
import helmet from "helmet";
6
-
import { pino } from "pino";
7
8
-
import { createDb, migrateToLatest } from "#/db";
9
-
import { env } from "#/env";
10
-
import { Ingester } from "#/firehose/ingester";
11
-
import errorHandler from "#/middleware/errorHandler";
12
-
import requestLogger from "#/middleware/requestLogger";
13
-
import { createRouter } from "#/routes";
14
-
import type { AppContext } from "./config";
15
16
export class Server {
17
constructor(
18
public app: express.Application,
19
public server: http.Server,
20
-
public ctx: AppContext
21
) {}
22
23
static async create() {
24
-
const { NODE_ENV, HOST, PORT } = env;
25
26
-
const logger = pino({ name: "server start" });
27
-
const db = createDb(":memory:");
28
-
await migrateToLatest(db);
29
-
const ingester = new Ingester(db);
30
-
ingester.start();
31
const ctx = {
32
db,
33
ingester,
34
logger,
35
-
};
36
37
-
const app: Express = express();
38
39
// Set the application to trust the reverse proxy
40
-
app.set("trust proxy", true);
41
42
// TODO: middleware for sqlite server
43
// TODO: middleware for OAuth
44
45
// Middlewares
46
-
app.use(express.json());
47
-
app.use(express.urlencoded({ extended: true }));
48
-
app.use(cors({ origin: env.CORS_ORIGIN, credentials: true }));
49
-
app.use(helmet());
50
51
// Request logging
52
-
app.use(requestLogger);
53
54
// Routes
55
-
const router = createRouter(ctx);
56
-
app.use(router);
57
58
// Error handlers
59
-
app.use(errorHandler());
60
61
-
const server = app.listen(env.PORT);
62
-
await events.once(server, "listening");
63
-
logger.info(`Server (${NODE_ENV}) running on port http://${HOST}:${PORT}`);
64
65
-
return new Server(app, server, ctx);
66
}
67
68
async close() {
69
-
this.ctx.logger.info("sigint received, shutting down");
70
-
this.ctx.ingester.destroy();
71
return new Promise<void>((resolve) => {
72
this.server.close(() => {
73
-
this.ctx.logger.info("server closed");
74
-
resolve();
75
-
});
76
-
});
77
}
78
}
···
1
+
import events from 'node:events'
2
+
import type http from 'node:http'
3
+
import cors from 'cors'
4
+
import express, { type Express } from 'express'
5
+
import helmet from 'helmet'
6
+
import { pino } from 'pino'
7
8
+
import { createDb, migrateToLatest } from '#/db'
9
+
import { env } from '#/env'
10
+
import { Ingester } from '#/firehose/ingester'
11
+
import errorHandler from '#/middleware/errorHandler'
12
+
import requestLogger from '#/middleware/requestLogger'
13
+
import { createRouter } from '#/routes'
14
+
import type { AppContext } from './config'
15
16
export class Server {
17
constructor(
18
public app: express.Application,
19
public server: http.Server,
20
+
public ctx: AppContext,
21
) {}
22
23
static async create() {
24
+
const { NODE_ENV, HOST, PORT } = env
25
26
+
const logger = pino({ name: 'server start' })
27
+
const db = createDb(':memory:')
28
+
await migrateToLatest(db)
29
+
const ingester = new Ingester(db)
30
+
ingester.start()
31
const ctx = {
32
db,
33
ingester,
34
logger,
35
+
}
36
37
+
const app: Express = express()
38
39
// Set the application to trust the reverse proxy
40
+
app.set('trust proxy', true)
41
42
// TODO: middleware for sqlite server
43
// TODO: middleware for OAuth
44
45
// Middlewares
46
+
app.use(express.json())
47
+
app.use(express.urlencoded({ extended: true }))
48
+
app.use(cors({ origin: env.CORS_ORIGIN, credentials: true }))
49
+
app.use(helmet())
50
51
// Request logging
52
+
app.use(requestLogger)
53
54
// Routes
55
+
const router = createRouter(ctx)
56
+
app.use(router)
57
58
// Error handlers
59
+
app.use(errorHandler())
60
61
+
const server = app.listen(env.PORT)
62
+
await events.once(server, 'listening')
63
+
logger.info(`Server (${NODE_ENV}) running on port http://${HOST}:${PORT}`)
64
65
+
return new Server(app, server, ctx)
66
}
67
68
async close() {
69
+
this.ctx.logger.info('sigint received, shutting down')
70
+
this.ctx.ingester.destroy()
71
return new Promise<void>((resolve) => {
72
this.server.close(() => {
73
+
this.ctx.logger.info('server closed')
74
+
resolve()
75
+
})
76
+
})
77
}
78
}