this repo has no description
woo
+371
-236
+1
.env.example
+1
.env.example
+2
package.json
+2
package.json
···
18
18
"@types/express": "^5.0.0",
19
19
"@types/pg": "^8.11.10",
20
20
"@types/terminal-kit": "^2.5.6",
21
+
"async-mutex": "^0.5.0",
21
22
"axios": "^1.7.7",
22
23
"big-json": "^3.2.0",
23
24
"cli-progress": "^3.12.0",
···
25
26
"express": "^4.21.1",
26
27
"ky": "^1.7.2",
27
28
"kysely": "^0.27.4",
29
+
"kysely-codegen": "^0.16.8",
28
30
"libsql": "^0.4.6",
29
31
"p-limit": "^6.1.0",
30
32
"pg": "^8.13.0",
+7
-3
src/constants.ts
+7
-3
src/constants.ts
···
1
+
import 'dotenv/config';
1
2
export const RELAY_URL = process.env.RELAY_URL!;
2
3
export const PLC_DB_PATH = process.env.PLC_DB_PATH!;
4
+
export const METRICS_PORT = parseInt(process.env.METRICS_PORT!, 10);
3
5
export const SQL_OUTPUT_FILE = 'dids_pds.jsonl';
4
6
export const HEALTH_CHECK_FILE = 'pds_health.json';
5
7
export const DATA_OUTPUT_FILE = 'bsky_data.jsonl';
6
8
export const PDS_HEALTH_CHECK_CONCURRENCY = 20;
7
-
export const PDS_DATA_FETCH_CONCURRENCY = 100;
8
-
export const DIDS_TO_PROCESS = 20000;
9
-
export const METRICS_PORT = 9501;
9
+
export const PDS_DATA_FETCH_CONCURRENCY = 150;
10
+
export const DIDS_TO_PROCESS = parseInt(process.argv[2], 10) || 10000;
11
+
export const BATCH_SIZE = 5000; // Number of records per batch
12
+
export const BATCH_TIMEOUT_MS = 1000; // 1 second
13
+
export const MAX_FLUSH_RETRIES = 5; // Maximum number of retry attempts for flushing
+4
-1
src/helpers.ts
+4
-1
src/helpers.ts
···
1
1
import ky from 'ky';
2
2
3
3
import { RELAY_URL } from './constants.js';
4
-
import logger from './logger.js';
5
4
import { ServerDescription } from './types.js';
6
5
7
6
export function sanitizePDSName(pds: string): string {
···
57
56
return false;
58
57
}
59
58
}
59
+
60
+
export function sanitizeTimestamp(timestamp: string): string {
61
+
return timestamp.startsWith('0000-') ? timestamp.replace('0000-', '0001-') : timestamp;
62
+
}
+17
-42
src/index.ts
+17
-42
src/index.ts
···
1
-
import 'dotenv/config';
2
-
3
-
import {
4
-
DIDS_TO_PROCESS,
5
-
} from './constants.js';
6
-
import logger from './logger.js';
1
+
import { DIDS_TO_PROCESS, METRICS_PORT } from './constants.js';
2
+
import { gracefulShutdown, registerShutdownHandlers } from './shutdown.js';
7
3
import { fetchAndDumpDidsPdses } from './stages/stage1.js';
8
-
import { processDidsAndFetchData } from './stages/stage3.js';
9
4
import { checkAllPDSHealth, selectAllDids } from './stages/stage2.js';
5
+
import { processDidsAndFetchData } from './stages/stage3.js';
6
+
import { DidAndPds } from './types.js';
7
+
import { startMetricsServer, stopMetricsServer } from './metrics.js';
10
8
9
+
async function main() {
10
+
// Register graceful shutdown handlers
11
+
registerShutdownHandlers();
11
12
13
+
// start metrics server
14
+
startMetricsServer(METRICS_PORT);
12
15
13
-
async function main() {
14
16
// stage 1
15
17
await fetchAndDumpDidsPdses();
16
18
17
19
// stage 2
18
20
const { groupedByPDS, pdsHealthStatus } = await checkAllPDSHealth();
19
-
const allDids: { did: string; pds: string; }[] = selectAllDids(groupedByPDS, pdsHealthStatus);
21
+
const allDids: DidAndPds[] = selectAllDids(groupedByPDS, pdsHealthStatus);
20
22
21
23
// stage 3
22
24
const didsToProcess = allDids.slice(0, DIDS_TO_PROCESS);
23
25
console.log(`Processing ${didsToProcess.length} DIDs`);
24
26
25
-
const fetchedData = await processDidsAndFetchData(didsToProcess);
27
+
await processDidsAndFetchData(didsToProcess);
28
+
29
+
await gracefulShutdown();
26
30
27
-
console.log(`Fetched data array length: ${fetchedData.length}`);
31
+
// console.log(`Fetched data array length: ${fetchedData.length}`);
28
32
}
29
33
30
-
main().catch((error: unknown) => {
34
+
main().catch(async (error: unknown) => {
31
35
console.error(`Unexpected error: ${error instanceof Error ? error.message : String(error)}`);
36
+
await gracefulShutdown();
32
37
process.exit(1);
33
38
});
34
-
35
-
let isShuttingDown = false;
36
-
37
-
38
-
async function shutdown() {
39
-
if (isShuttingDown) {
40
-
console.log('Shutdown called but one is already in progress.');
41
-
return;
42
-
}
43
-
44
-
isShuttingDown = true;
45
-
46
-
console.log('Shutting down gracefully...');
47
-
48
-
process.exit(0);
49
-
}
50
-
51
-
process.on('SIGINT', () => {
52
-
shutdown().catch((error: unknown) => {
53
-
console.error(`Shutdown failed: ${(error as Error).message}`);
54
-
process.exit(1);
55
-
});
56
-
});
57
-
58
-
process.on('SIGTERM', () => {
59
-
shutdown().catch((error: unknown) => {
60
-
console.error(`Shutdown failed: ${(error as Error).message}`);
61
-
process.exit(1);
62
-
});
63
-
});
+34
-8
src/metrics.ts
+34
-8
src/metrics.ts
···
1
-
// import { monitorPgPool } from '@christiangalsterer/node-postgres-prometheus-exporter';
1
+
import { monitorPgPool } from '@christiangalsterer/node-postgres-prometheus-exporter';
2
2
import express from 'express';
3
-
import { Registry, collectDefaultMetrics } from 'prom-client';
3
+
import { Gauge, Registry, collectDefaultMetrics } from 'prom-client';
4
4
5
-
import logger from './logger.js';
6
-
// import { pool } from './postgres.js';
5
+
import { pool } from './postgres.js';
6
+
import { Server } from 'http';
7
7
8
8
const register = new Registry();
9
9
collectDefaultMetrics({ register });
10
10
11
-
// monitorPgPool(pool, register);
11
+
export const concurrentPostgresInserts = new Gauge({
12
+
name: 'bluesky_concurrent_postgres_inserts',
13
+
help: 'Number of concurrent Postgres inserts',
14
+
registers: [register],
15
+
});
16
+
17
+
monitorPgPool(pool, register);
12
18
13
19
const app = express();
14
20
···
25
31
});
26
32
});
27
33
34
+
let metricsServer: Server;
35
+
28
36
export const startMetricsServer = (port: number, host = '127.0.0.1') => {
29
-
const server = app.listen(port, host, () => {
37
+
metricsServer = app.listen(port, host, () => {
30
38
console.log(`Metrics server listening on port ${port}`);
31
39
});
32
40
33
-
server.on('close', () => {
41
+
metricsServer.on('close', () => {
34
42
console.log('Metrics server closed.');
35
43
});
36
44
37
-
return server;
45
+
return metricsServer;
38
46
};
47
+
48
+
export function stopMetricsServer(): Promise<void> {
49
+
return new Promise((resolve, reject) => {
50
+
if (metricsServer) {
51
+
metricsServer.close((err) => {
52
+
if (err) {
53
+
console.error('Error shutting down metrics server:', err);
54
+
reject(err);
55
+
} else {
56
+
console.log('Metrics server shut down successfully.');
57
+
resolve();
58
+
}
59
+
});
60
+
} else {
61
+
resolve();
62
+
}
63
+
});
64
+
}
+50
src/migrations/001-create.ts
+50
src/migrations/001-create.ts
···
1
+
import 'dotenv/config';
2
+
import pg from 'pg';
3
+
4
+
const { Client } = pg;
5
+
6
+
const client = new Client({
7
+
connectionString: process.env.DATABASE_URL!,
8
+
});
9
+
10
+
await client.connect();
11
+
12
+
export async function createTables() {
13
+
await client.query(`
14
+
CREATE TABLE IF NOT EXISTS posts (
15
+
id BIGSERIAL PRIMARY KEY,
16
+
cid TEXT NOT NULL, -- ~64 characters
17
+
did TEXT NOT NULL, -- ~32 characters
18
+
rkey TEXT NOT NULL, -- ~13 characters
19
+
has_emojis BOOLEAN NOT NULL DEFAULT FALSE,
20
+
langs TEXT[] NOT NULL DEFAULT '{}',
21
+
text TEXT,
22
+
created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT (now() at time zone 'utc')
23
+
);
24
+
25
+
CREATE TABLE IF NOT EXISTS emojis (
26
+
id BIGSERIAL PRIMARY KEY,
27
+
post_id BIGINT DEFAULT NULL,
28
+
profile_id BIGINT DEFAULT NULL,
29
+
emoji TEXT NOT NULL,
30
+
lang TEXT NOT NULL, -- 2 or 5 characters
31
+
created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT (now() at time zone 'utc')
32
+
);
33
+
34
+
CREATE TABLE IF NOT EXISTS profiles (
35
+
id BIGSERIAL PRIMARY KEY,
36
+
cid TEXT NOT NULL, -- ~64 characters
37
+
did TEXT NOT NULL, -- ~32 characters
38
+
rkey TEXT NOT NULL, -- ~13 characters
39
+
description TEXT,
40
+
display_name TEXT,
41
+
created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT (now() at time zone 'utc')
42
+
);
43
+
`);
44
+
}
45
+
46
+
createTables()
47
+
.catch((e: unknown) => {
48
+
console.error(e);
49
+
})
50
+
.finally(() => void client.end());
+5
-1
src/postgres.ts
+5
-1
src/postgres.ts
+159
src/postgresBatchQueue.ts
+159
src/postgresBatchQueue.ts
···
1
+
// src/postgresBatchQueue.ts
2
+
3
+
import { Mutex } from 'async-mutex';
4
+
import { db, closeDatabase } from './postgres.js';
5
+
import { concurrentPostgresInserts } from './metrics.js';
6
+
import { PostData } from './types.js';
7
+
import { BATCH_SIZE, BATCH_TIMEOUT_MS, MAX_FLUSH_RETRIES } from './constants.js';
8
+
9
+
export class PostgresBatchQueue {
10
+
private queue: PostData[] = [];
11
+
private mutex = new Mutex();
12
+
private batchSize: number;
13
+
private batchTimeoutMs: number;
14
+
private batchTimer: NodeJS.Timeout | null = null;
15
+
private isShuttingDown = false;
16
+
17
+
constructor(batchSize: number, batchTimeoutMs: number) {
18
+
this.batchSize = batchSize;
19
+
this.batchTimeoutMs = batchTimeoutMs;
20
+
}
21
+
22
+
/**
23
+
* Adds a PostData item to the queue and triggers flush if necessary.
24
+
* @param data PostData to enqueue
25
+
*/
26
+
public async enqueue(data: PostData): Promise<void> {
27
+
if (this.isShuttingDown) {
28
+
throw new Error('Cannot enqueue data, the queue is shutting down.');
29
+
}
30
+
31
+
let shouldFlush = false;
32
+
33
+
await this.mutex.runExclusive(() => {
34
+
this.queue.push(data);
35
+
36
+
if (this.queue.length >= this.batchSize) {
37
+
shouldFlush = true;
38
+
} else if (!this.batchTimer) {
39
+
this.scheduleFlush();
40
+
}
41
+
});
42
+
43
+
if (shouldFlush) {
44
+
await this.flushQueue();
45
+
}
46
+
}
47
+
48
+
/**
49
+
* Schedules a flush after the specified timeout.
50
+
*/
51
+
private scheduleFlush(): void {
52
+
this.batchTimer = setTimeout(() => {
53
+
this.flushQueue().catch((err) => {
54
+
console.error(`Scheduled flush error: ${(err as Error).message}`);
55
+
});
56
+
}, this.batchTimeoutMs);
57
+
}
58
+
59
+
/**
60
+
* Flushes the current queue to PostgreSQL.
61
+
*/
62
+
private async flushQueue(): Promise<void> {
63
+
// Clear the existing timer
64
+
if (this.batchTimer) {
65
+
clearTimeout(this.batchTimer);
66
+
this.batchTimer = null;
67
+
}
68
+
69
+
let currentBatch: PostData[] = [];
70
+
71
+
await this.mutex.runExclusive(() => {
72
+
if (this.queue.length === 0) {
73
+
return;
74
+
}
75
+
currentBatch = this.queue.splice(0, this.batchSize);
76
+
});
77
+
78
+
if (currentBatch.length === 0) {
79
+
return;
80
+
}
81
+
82
+
concurrentPostgresInserts.inc();
83
+
84
+
try {
85
+
await this.attemptFlush(currentBatch);
86
+
process.stdout.write('.');
87
+
} catch (error) {
88
+
console.error(`Error flushing PostgreSQL batch: ${(error as Error).message}`);
89
+
// Re-add the failed batch back for retry
90
+
await this.mutex.runExclusive(() => {
91
+
this.queue = currentBatch.concat(this.queue);
92
+
});
93
+
} finally {
94
+
concurrentPostgresInserts.dec();
95
+
}
96
+
}
97
+
98
+
/**
99
+
* Attempts to flush the batch with retry logic.
100
+
* @param batch The batch of PostData to flush
101
+
*/
102
+
private async attemptFlush(batch: PostData[]): Promise<void> {
103
+
let attempt = 0;
104
+
let success = false;
105
+
106
+
while (attempt < MAX_FLUSH_RETRIES && !success) {
107
+
try {
108
+
await db.transaction().execute(async (tx) => {
109
+
await tx
110
+
.insertInto('posts') // Ensure 'posts' is your table name
111
+
.values(
112
+
batch.map((post) => ({
113
+
cid: post.cid,
114
+
did: post.did,
115
+
rkey: post.rkey,
116
+
has_emojis: post.hasEmojis,
117
+
langs: post.langs,
118
+
text: post.post,
119
+
created_at: post.createdAt,
120
+
})),
121
+
)
122
+
.execute();
123
+
});
124
+
125
+
success = true;
126
+
} catch (error) {
127
+
attempt++;
128
+
console.error(`Flush attempt ${attempt} failed: ${(error as Error).message}`);
129
+
130
+
if (attempt < MAX_FLUSH_RETRIES) {
131
+
const backoffTime = 2 ** attempt * 1000; // Exponential backoff
132
+
console.log(`Retrying in ${backoffTime} ms...`);
133
+
await new Promise((resolve) => setTimeout(resolve, backoffTime));
134
+
} else {
135
+
console.error('Max retries reached. Re-queueing the batch.');
136
+
throw error; // Let the caller handle re-queueing
137
+
}
138
+
}
139
+
}
140
+
}
141
+
142
+
/**
143
+
* Gracefully shuts down the queue by flushing remaining items.
144
+
*/
145
+
public async shutdown(): Promise<void> {
146
+
this.isShuttingDown = true;
147
+
if (this.batchTimer) {
148
+
clearTimeout(this.batchTimer);
149
+
this.batchTimer = null;
150
+
}
151
+
await this.flushQueue();
152
+
console.log('Flushed all remaining items.');
153
+
await closeDatabase();
154
+
console.log('Database connections closed.');
155
+
}
156
+
}
157
+
158
+
// Instantiate the queue
159
+
export const postgresBatchQueue = new PostgresBatchQueue(BATCH_SIZE, BATCH_TIMEOUT_MS);
+13
src/schema.d.ts
+13
src/schema.d.ts
···
9
9
? ColumnType<S, I | undefined, U>
10
10
: ColumnType<T, T | undefined, T>;
11
11
12
+
export type Int8 = ColumnType<string, bigint | number | string>;
13
+
12
14
export type Timestamp = ColumnType<Date, Date | string>;
13
15
14
16
export interface Emojis {
···
30
32
text: string | null;
31
33
}
32
34
35
+
export interface Profiles {
36
+
cid: string;
37
+
created_at: Generated<Timestamp>;
38
+
description: string | null;
39
+
did: string;
40
+
display_name: string | null;
41
+
id: Generated<Int8>;
42
+
rkey: string;
43
+
}
44
+
33
45
export interface DB {
34
46
emojis: Emojis;
35
47
posts: Posts;
48
+
profiles: Profiles;
36
49
}
+34
src/shutdown.ts
+34
src/shutdown.ts
···
1
+
// src/shutdown.ts
2
+
3
+
import { postgresBatchQueue } from './postgresBatchQueue.js';
4
+
import { closeDatabase } from './postgres.js';
5
+
import { stopMetricsServer } from './metrics.js';
6
+
7
+
export async function gracefulShutdown(): Promise<void> {
8
+
console.log('Initiating graceful shutdown...');
9
+
try {
10
+
await stopMetricsServer();
11
+
await postgresBatchQueue.shutdown();
12
+
console.log('All pending batches have been flushed.');
13
+
await closeDatabase();
14
+
console.log('Database connections closed.');
15
+
process.exit(0);
16
+
} catch (err) {
17
+
console.error(`Error during shutdown: ${(err as Error).message}`);
18
+
process.exit(1);
19
+
}
20
+
}
21
+
22
+
// Register shutdown handlers
23
+
export function registerShutdownHandlers() {
24
+
process.on('SIGINT', gracefulShutdown);
25
+
process.on('SIGTERM', gracefulShutdown);
26
+
process.on('uncaughtException', async (err) => {
27
+
console.error('Uncaught Exception:', err);
28
+
await gracefulShutdown();
29
+
});
30
+
process.on('unhandledRejection', async (reason, promise) => {
31
+
console.error('Unhandled Rejection at:', promise, 'reason:', reason);
32
+
await gracefulShutdown();
33
+
});
34
+
}
-2
src/stages/stage1.ts
-2
src/stages/stage1.ts
···
1
-
import 'dotenv/config';
2
1
import Database from 'libsql';
3
2
import fs from 'node:fs/promises';
4
3
import { performance } from 'node:perf_hooks';
5
4
6
5
import { PLC_DB_PATH, RELAY_URL, SQL_OUTPUT_FILE } from '../constants.js';
7
-
import logger from '../logger.js';
8
6
import { DidAndPds } from '../types.js';
9
7
10
8
const didDb = new Database(PLC_DB_PATH);
-1
src/stages/stage2.ts
-1
src/stages/stage2.ts
+35
-178
src/stages/stage3.ts
+35
-178
src/stages/stage3.ts
···
2
2
import pLimit from 'p-limit';
3
3
4
4
import { PDS_DATA_FETCH_CONCURRENCY } from '../constants.js';
5
-
import logger from '../logger.js';
6
-
import { db } from '../postgres.js';
7
-
import { BskyData, BskyPost, BskyPostData } from '../types.js';
8
-
9
-
const BATCH_SIZE = 1000;
10
-
const BATCH_TIMEOUT_MS = 1000;
11
-
12
-
interface PostData {
13
-
cid: string;
14
-
did: string;
15
-
rkey: string;
16
-
hasEmojis: boolean;
17
-
langs: string[];
18
-
post: string;
19
-
createdAt: string;
20
-
}
21
-
22
-
let postBatch: PostData[] = [];
23
-
let postBatchCount = 0;
24
-
let isBatching = false;
25
-
let batchTimer: NodeJS.Timeout | null = null;
26
-
27
-
let isShuttingDown = false;
28
-
const ongoingHandleCreates = 0;
29
-
let shutdownPromise: Promise<void> | null = null;
30
-
31
-
function createShutdownPromise(): Promise<void> {
32
-
return new Promise<void>((resolve) => {
33
-
const checkCompletion = setInterval(() => {
34
-
console.log(`Shutting down, ongoing handleCreates: ${ongoingHandleCreates}`);
35
-
if (isShuttingDown && ongoingHandleCreates === 0) {
36
-
console.log('All ongoing handleCreate operations have finished.');
37
-
clearInterval(checkCompletion);
38
-
resolve();
39
-
}
40
-
}, 50);
41
-
});
42
-
}
5
+
import { sanitizeTimestamp } from '../helpers.js';
6
+
import { postgresBatchQueue } from '../postgresBatchQueue.js';
7
+
import { BskyData, BskyPost, BskyPostData, DidAndPds, PostData } from '../types.js';
43
8
44
-
export function initiateShutdown(): Promise<void> {
45
-
if (!shutdownPromise) {
46
-
isShuttingDown = true;
47
-
shutdownPromise = createShutdownPromise();
48
-
}
49
-
return shutdownPromise;
50
-
}
51
-
52
-
export async function flushPostgresBatch() {
53
-
if (postBatch.length === 0) {
54
-
isBatching = false;
55
-
return;
56
-
}
57
-
58
-
const currentBatch = [...postBatch];
59
-
postBatch = [];
60
-
isBatching = false;
61
-
62
-
// concurrentPostgresInserts.inc();
63
-
64
-
try {
65
-
await db.transaction().execute(async (tx) => {
66
-
// Bulk insert posts
67
-
const insertedPosts = await tx
68
-
.insertInto('posts')
69
-
.values(
70
-
currentBatch.map((post) => ({
71
-
cid: post.cid,
72
-
did: post.did,
73
-
rkey: post.rkey,
74
-
has_emojis: post.hasEmojis,
75
-
langs: post.langs,
76
-
text: post.post,
77
-
created_at: post.createdAt,
78
-
})),
79
-
)
80
-
// .returning(['id', 'cid', 'did', 'rkey'])
81
-
.execute();
82
-
83
-
// // Map composite key to id
84
-
// const compositeKeyToIdMap = new Map<string, number>();
85
-
// insertedPosts.forEach((post) => {
86
-
// const compositeKey = `${post.cid}-${post.did}-${post.rkey}`;
87
-
// compositeKeyToIdMap.set(compositeKey, post.id);
88
-
// });
89
-
90
-
// // Prepare bulk insert for emojis
91
-
// const emojiInserts: { post_id: number; emoji: string; lang: string }[] = [];
92
-
// currentBatch.forEach((post) => {
93
-
// if (post.hasEmojis) {
94
-
// const compositeKey = `${post.cid}-${post.did}-${post.rkey}`;
95
-
// const postId = compositeKeyToIdMap.get(compositeKey);
96
-
// if (postId) {
97
-
// post.emojis.forEach((emoji) => {
98
-
// post.langs.forEach((lang) => {
99
-
// emojiInserts.push({
100
-
// post_id: postId,
101
-
// emoji: emoji,
102
-
// lang: lang,
103
-
// });
104
-
// });
105
-
// });
106
-
// }
107
-
// }
108
-
// });
109
-
110
-
// if (emojiInserts.length > 0) {
111
-
// await tx.insertInto('emojis').values(emojiInserts).execute();
112
-
// }
113
-
});
114
-
115
-
// concurrentPostgresInserts.dec();
116
-
} catch (error) {
117
-
console.error(`Error flushing PostgreSQL batch: ${(error as Error).message}`);
118
-
// Optionally, you can re-add the failed batch back to `postBatch` for retry
119
-
postBatch = currentBatch.concat(postBatch);
120
-
}
121
-
}
122
-
123
-
/**
124
-
* Schedule a batch flush after a timeout.
125
-
*/
126
-
function scheduleBatchFlush() {
127
-
if (batchTimer) {
128
-
return;
129
-
}
130
-
batchTimer = setTimeout(() => {
131
-
batchTimer = null;
132
-
void flushPostgresBatch();
133
-
}, BATCH_TIMEOUT_MS);
134
-
}
135
-
136
-
export async function processDidsAndFetchData(dids: { did: string; pds: string }[]) {
9
+
export async function processDidsAndFetchData(dids: DidAndPds[]): Promise<void> {
137
10
const limit = pLimit(PDS_DATA_FETCH_CONCURRENCY);
138
-
const fetchedData: BskyData[] = [];
139
11
let successfulRequests = 0;
140
12
let unsuccessfulRequests = 0;
141
13
let successfulDids = 0;
···
149
21
{ did, pds },
150
22
{
151
23
responseType: 'stream',
152
-
timeout: 60000,
24
+
timeout: 30 * 60 * 1000, // 30 minutes
153
25
},
154
26
);
155
27
···
157
29
158
30
await new Promise<void>((resolve, reject) => {
159
31
let buffer = '';
160
-
let didSucceeded = false;
161
32
162
33
// eslint-disable-next-line @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-member-access
163
34
res.data.on('data', (chunk: Buffer) => {
···
169
40
if (line.trim()) {
170
41
try {
171
42
const json = JSON.parse(line) as BskyData;
172
-
postBatchCount++;
173
-
if (postBatchCount % 1000 === 0) {
174
-
process.stdout.write('.');
175
-
void flushPostgresBatch();
176
-
}
177
-
// console.log('json', json);
178
-
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
179
-
if (json) {
180
-
for (const [k, v] of Object.entries(json)) {
181
-
if (k.includes('app.bsky.feed.post')) {
182
-
const post = v as BskyPost;
183
-
const postData = (post.value as unknown) as BskyPostData;
184
-
postBatch.push({
185
-
cid: postData.cid,
186
-
did: did,
187
-
rkey: k.split('/')[1],
188
-
hasEmojis: false,
189
-
langs: postData.langs,
190
-
post: postData.text,
191
-
createdAt: postData.createdAt,
192
-
});
193
-
}
43
+
for (const [k, v] of Object.entries(json)) {
44
+
if (k.includes('app.bsky.feed.post')) {
45
+
const post = v as BskyPost;
46
+
const postData = post.value as unknown as BskyPostData;
47
+
const rkeyParts = k.split('/');
48
+
const rkey = rkeyParts.length > 1 ? rkeyParts[1] : k;
49
+
const sanitizedCreatedAt = sanitizeTimestamp(postData.createdAt);
50
+
const data: PostData = {
51
+
cid: postData.cid,
52
+
did: did,
53
+
rkey: rkey,
54
+
hasEmojis: false,
55
+
langs: postData.langs,
56
+
post: postData.text,
57
+
createdAt: sanitizedCreatedAt,
58
+
};
59
+
postgresBatchQueue.enqueue(data).catch((err: unknown) => {
60
+
console.error(`Enqueue error for DID ${did}: ${(err as Error).message}`);
61
+
});
194
62
}
195
63
}
196
-
didSucceeded = true;
197
64
} catch (err) {
198
65
console.error(`JSON parse error for DID ${did}: ${(err as Error).message}`);
199
66
}
···
204
71
205
72
// eslint-disable-next-line @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-member-access
206
73
res.data.on('end', () => {
207
-
scheduleBatchFlush();
208
74
if (buffer.trim()) {
209
75
try {
210
76
const json = JSON.parse(buffer) as BskyData;
211
-
console.log('OMG WE GOT HERE OOPS NEED TO HANDLE THIS');
212
-
console.log(json);
213
-
didSucceeded = true;
77
+
console.dir(json, { depth: null });
78
+
if (Object.keys(json).length > 0) {
79
+
throw new Error('JSON is not empty');
80
+
}
214
81
} catch (err) {
215
82
console.error(`JSON parse error at stream end for DID ${did}: ${(err as Error).message}`);
216
83
}
217
84
}
218
-
if (didSucceeded) {
219
-
successfulDids++;
220
-
} else {
221
-
failedDids++;
85
+
successfulDids++;
86
+
if (successfulDids % 100 === 0) {
87
+
process.stdout.write('#');
222
88
}
223
89
resolve();
224
90
});
···
227
93
res.data.on('error', (err: Error) => {
228
94
console.error(`Stream error for DID ${did}: ${err.message}`);
229
95
failedDids++;
96
+
if (failedDids % 100 === 0) {
97
+
process.stdout.write('*');
98
+
}
230
99
reject(err);
231
100
});
232
101
});
233
-
return;
234
102
} catch (error) {
235
-
process.stdout.write('!');
236
-
// this just means the user doesn't exist anymore for whatever reason
237
103
if (!(error as Error).message.includes('Request failed with status code 502')) {
238
104
console.error(`Error with DID ${did}: ${(error as Error).message}`);
239
105
}
···
244
110
);
245
111
246
112
await Promise.all(tasks);
247
-
console.log(`Fetched data for ${fetchedData.length} DIDs.`);
113
+
console.log(`Processed DIDs.`);
248
114
console.log(`Successful requests: ${successfulRequests}, Unsuccessful requests: ${unsuccessfulRequests}`);
249
115
console.log(`Successful DIDs: ${successfulDids}, Failed DIDs: ${failedDids}`);
250
-
251
-
// const writeFile = await fs.open(DATA_OUTPUT_FILE, 'w');
252
-
// const writeStream = writeFile.createWriteStream();
253
-
// for (const data of fetchedData) {
254
-
// writeStream.write(JSON.stringify(data) + '\n');
255
-
// }
256
-
// writeStream.close();
257
-
// console.log(`Streamed fetched data to ${DATA_OUTPUT_FILE}`);
258
-
return fetchedData;
259
116
}