this repo has no description
we don't need no database!
+4
-199
-1
.env.example
-1
.env.example
-109
src/db.ts
-109
src/db.ts
···
1
-
import { eq } from 'drizzle-orm';
2
-
import { drizzle } from 'drizzle-orm/node-postgres';
3
-
import * as pg from 'pg';
4
-
import dotenv from 'dotenv';
5
-
6
-
import logger from './logger.js';
7
-
import * as schema from './schema.js';
8
-
import { /*cursor, */languages, posts } from './schema.js';
9
-
10
-
dotenv.config();
11
-
12
-
const { Pool } = pg.default;
13
-
14
-
export const pool = new Pool({
15
-
connectionString: process.env.DATABASE_URL,
16
-
});
17
-
18
-
export const db = drizzle(pool, { schema });
19
-
20
-
// export async function getLastCursor(): Promise<number> {
21
-
// logger.debug('Getting last cursor...');
22
-
// const result = await db.select({ lastCursor: cursor.lastCursor }).from(cursor).where(eq(cursor.id, 1));
23
-
// if (result.length === 0) {
24
-
// logger.info('No cursor found, initializing with current epoch in microseconds...');
25
-
// const currentEpochMicroseconds = BigInt(Date.now()) * 1000n;
26
-
// await db
27
-
// .insert(cursor)
28
-
// .values({
29
-
// id: 1,
30
-
// lastCursor: Number(currentEpochMicroseconds),
31
-
// })
32
-
// .execute();
33
-
// logger.info(`Initialized cursor with value: ${currentEpochMicroseconds}`);
34
-
// return Number(currentEpochMicroseconds);
35
-
// }
36
-
// logger.info(`Returning cursor from database: ${result[0].lastCursor}`);
37
-
// return result[0].lastCursor!;
38
-
// }
39
-
40
-
// export async function updateLastCursor(newCursor: number): Promise<void> {
41
-
// try {
42
-
// await db.update(cursor).set({ lastCursor: newCursor }).where(eq(cursor.id, 1));
43
-
// logger.info(`Updated last cursor to ${newCursor}`);
44
-
// } catch (error: unknown) {
45
-
// logger.error(`Error updating cursor: ${(error as Error).message}`);
46
-
// }
47
-
// }
48
-
49
-
export async function savePost(post: {
50
-
id: string;
51
-
created_at: Date;
52
-
langs: Set<string>;
53
-
did: string;
54
-
rkey: string;
55
-
cursor: number;
56
-
}) {
57
-
try {
58
-
await db.transaction(async (tx) => {
59
-
await tx
60
-
.insert(posts)
61
-
.values({
62
-
id: post.id,
63
-
createdAt: post.created_at,
64
-
did: post.did,
65
-
rkey: post.rkey,
66
-
cursor: post.cursor,
67
-
})
68
-
.onConflictDoNothing();
69
-
70
-
const languageEntries = Array.from(post.langs).map((lang) => ({
71
-
postId: post.id,
72
-
language: lang,
73
-
}));
74
-
75
-
if (languageEntries.length > 0) {
76
-
await tx.insert(languages).values(languageEntries);
77
-
}
78
-
79
-
logger.debug(`Saved/Updated post ${post.id}`);
80
-
});
81
-
} catch (error) {
82
-
logger.error(`Database insertion/update error: ${(error as Error).message}`, { post });
83
-
}
84
-
}
85
-
86
-
export async function deletePost(postId: string): Promise<boolean> {
87
-
try {
88
-
const result = await db.delete(posts).where(eq(posts.id, postId)).returning();
89
-
if (result.length > 0) {
90
-
logger.info(`Deleted post ${postId}`);
91
-
return true;
92
-
} else {
93
-
logger.debug(`Attempted to delete non-existent post ${postId}`);
94
-
return false;
95
-
}
96
-
} catch (error) {
97
-
logger.error(`Error deleting post: ${(error as Error).message}`, { postId });
98
-
return false;
99
-
}
100
-
}
101
-
102
-
export async function closeDatabase(): Promise<void> {
103
-
try {
104
-
await pool.end();
105
-
logger.info('Database connection closed.');
106
-
} catch (error) {
107
-
logger.error(`Error closing database: ${(error as Error).message}`);
108
-
}
109
-
}
+3
-48
src/index.ts
+3
-48
src/index.ts
···
2
2
import dotenv from 'dotenv';
3
3
import process from 'process';
4
4
5
-
// import { closeDatabase, /*deletePost,*/ /*getLastCursor,*/ /*savePost, /*updateLastCursor */ } from './db.js';
6
5
import logger from './logger.js';
7
6
import { decrementPostsCount, incrementErrors, incrementMetrics, incrementPostsCount } from './metrics.js';
8
7
import { app } from './web.js';
···
11
10
12
11
const FIREHOSE_URL = process.env.FIREHOSE_URL ?? 'wss://jetstream.atproto.tools/subscribe'; // default to Jaz's Jetstream instance
13
12
const PORT = parseInt(process.env.PORT ?? '9201', 10);
14
-
// const CURSOR_UPDATE_INTERVAL_MS = 10 * 1000;
15
13
16
-
// let latestCursor = await getLastCursor();
17
-
// logger.info(`Initial cursor set to: ${latestCursor}`);
18
-
// let cursorUpdateInterval: NodeJS.Timeout | null = null;
19
-
20
-
// function initializeCursorUpdate() {
21
-
// cursorUpdateInterval = setInterval(() => {
22
-
// if (latestCursor > 0) {
23
-
// updateLastCursor(latestCursor)
24
-
// .then(() => {
25
-
// logger.info(`Cursor updated to ${latestCursor} at ${new Date().toISOString()}`);
26
-
// })
27
-
// .catch((error: unknown) => {
28
-
// logger.error(`Error updating cursor: ${(error as Error).message}`);
29
-
// });
30
-
// }
31
-
// }, CURSOR_UPDATE_INTERVAL_MS);
32
-
// }
33
-
34
-
/*async*/ function handleCreate(event: CommitCreateEvent<'app.bsky.feed.post'>) {
14
+
function handleCreate(event: CommitCreateEvent<'app.bsky.feed.post'>) {
35
15
const { commit } = event;
36
16
37
17
if (!commit.rkey) return;
···
55
35
rkey: rkey,
56
36
cursor: event.time_us,
57
37
};
58
-
// await savePost(post);
59
38
incrementMetrics(post.langs);
60
39
incrementPostsCount();
61
-
// if (event.time_us > latestCursor) {
62
-
// latestCursor = event.time_us;
63
-
// }
64
40
} catch (error) {
65
41
logger.error(`Error parsing record in "create" commit: ${(error as Error).message}`, { commit, record });
66
42
logger.error(`Malformed record data: ${JSON.stringify(record)}`);
···
68
44
}
69
45
}
70
46
71
-
/*async*/ function handleDelete(event: CommitEvent<'app.bsky.feed.post'>) {
47
+
function handleDelete(event: CommitEvent<'app.bsky.feed.post'>) {
72
48
const { commit } = event;
73
49
74
50
if (!commit.rkey) return;
75
51
76
-
try {
77
-
// const postId = `${event.did}:${commit.rkey}`;
78
-
// const success = await deletePost(postId);
79
-
// if (success) {
80
-
decrementPostsCount();
81
-
// }
82
-
// if (event.time_us > latestCursor) {
83
-
// latestCursor = event.time_us;
84
-
// }
85
-
} catch (error) {
86
-
// logger.error(`Error deleting post: ${(error as Error).message}`, { rkey: commit.rkey });
87
-
incrementErrors();
88
-
}
52
+
decrementPostsCount();
89
53
}
90
54
91
55
const server = app.listen(PORT, '127.0.0.1', () => {
···
95
59
const jetstream = new Jetstream({
96
60
wantedCollections: ['app.bsky.feed.post'],
97
61
endpoint: FIREHOSE_URL,
98
-
// cursor: latestCursor.toString(),
99
62
});
100
63
101
64
jetstream.start();
102
65
103
66
jetstream.on('open', () => {
104
67
logger.info('Connected to Jetstream firehose.');
105
-
// if (!cursorUpdateInterval) {
106
-
// initializeCursorUpdate();
107
-
// }
108
68
});
109
69
110
70
jetstream.on('close', () => {
···
127
87
function shutdown() {
128
88
logger.info('Shutting down gracefully...');
129
89
130
-
// if (cursorUpdateInterval) {
131
-
// clearInterval(cursorUpdateInterval);
132
-
// }
133
-
134
90
server.close(() => {
135
91
logger.info('HTTP server closed.');
136
92
137
93
jetstream.close();
138
-
// void closeDatabase();
139
94
process.exit(0);
140
95
});
141
96
+1
-1
src/metrics.ts
+1
-1
src/metrics.ts
-8
src/migrate.ts
-8
src/migrate.ts
-32
src/schema.ts
-32
src/schema.ts
···
1
-
import { bigint, index, pgTable, serial, text, timestamp } from 'drizzle-orm/pg-core';
2
-
3
-
export const posts = pgTable('posts', {
4
-
id: text('id').primaryKey(),
5
-
createdAt: timestamp('created_at', { mode: 'date', withTimezone: true, precision: 6 }),
6
-
did: text('did'),
7
-
rkey: text('rkey'),
8
-
cursor: bigint('cursor', { mode: 'number' }),
9
-
}, (table) => {
10
-
return {
11
-
createdAtIndex: index('created_at_idx').on(table.createdAt),
12
-
};
13
-
});
14
-
15
-
export const languages = pgTable(
16
-
'languages',
17
-
{
18
-
postId: text('post_id').references(() => posts.id, { onDelete: 'cascade' }),
19
-
language: text('language'),
20
-
},
21
-
(table) => {
22
-
return {
23
-
postIdIdx: index('post_id_idx').on(table.postId),
24
-
languageIdx: index('language_idx').on(table.language),
25
-
};
26
-
},
27
-
);
28
-
29
-
export const cursor = pgTable('cursor', {
30
-
id: serial('id').primaryKey(),
31
-
lastCursor: bigint('last_cursor', { mode: 'number' }),
32
-
});