this repo has no description
fixes
+28
-57
+28
-25
src/main.ts
+28
-25
src/main.ts
···
9
9
let cursor = 0;
10
10
let cursorUpdateInterval: NodeJS.Timeout;
11
11
12
+
function epochUsToDateTime(cursor: number): string {
13
+
return new Date(cursor / 1000).toISOString();
14
+
}
15
+
12
16
try {
13
17
logger.info('Trying to read cursor from cursor.txt...');
14
18
cursor = Number(fs.readFileSync('cursor.txt', 'utf8'));
15
-
logger.info(`Cursor found: ${cursor} (${new Date(cursor / 1000).toISOString()})`);
19
+
logger.info(`Cursor found: ${cursor} (${epochUsToDateTime(cursor)})`);
16
20
} catch (error) {
17
21
if (error instanceof Error && 'code' in error && error.code === 'ENOENT') {
18
-
logger.info(
19
-
`Cursor not found in cursor.txt, setting cursor to: ${cursor} (${new Date(cursor / 1000).toISOString()})`,
20
-
);
22
+
logger.info(`Cursor not found in cursor.txt, setting cursor to: ${cursor} (${epochUsToDateTime(cursor)})`);
21
23
fs.writeFileSync('cursor.txt', cursor.toString(), 'utf8');
22
24
} else {
23
25
logger.error(error);
···
32
34
});
33
35
34
36
jetstream.on('open', () => {
35
-
logger.info(`Connected to Jetstream at ${FIREHOSE_URL}`);
37
+
logger.info(
38
+
`Connected to Jetstream at ${FIREHOSE_URL} with cursor ${jetstream.cursor} (${epochUsToDateTime(jetstream.cursor!)})`,
39
+
);
36
40
cursorUpdateInterval = setInterval(() => {
37
-
logger.info(`Cursor updated to: ${cursor} (${new Date(cursor / 1000).toISOString()})`);
38
-
fs.writeFile('cursor.txt', cursor.toString(), (err) => {
39
-
if (err) logger.error(err);
40
-
});
41
+
if (jetstream.cursor) {
42
+
logger.info(`Cursor updated to: ${jetstream.cursor} (${epochUsToDateTime(jetstream.cursor)})`);
43
+
fs.writeFile('cursor.txt', jetstream.cursor.toString(), (err) => {
44
+
if (err) logger.error(err);
45
+
});
46
+
}
41
47
}, CURSOR_UPDATE_INTERVAL);
42
48
});
43
49
···
51
57
});
52
58
53
59
jetstream.onCreate(WANTED_COLLECTION, (event: CommitCreateEvent<typeof WANTED_COLLECTION>) => {
54
-
cursor = event.time_us;
55
-
60
+
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
56
61
if (event.commit?.record?.subject?.uri?.includes(DID)) {
57
-
label(event.did, event.commit.record.subject.uri.split('/').pop()).catch((error: unknown) => {
58
-
logger.error(`Unexpected error labeling ${event.did}:`);
59
-
logger.error(error);
62
+
label(event.did, event.commit.record.subject.uri.split('/').pop()!).catch((error: unknown) => {
63
+
logger.error(`Unexpected error labeling ${event.did}: ${error}`);
60
64
});
61
65
}
62
66
});
···
73
77
74
78
jetstream.start();
75
79
80
+
// this doesn't work properly, need to research why
76
81
function shutdown() {
77
-
setTimeout(() => {
78
-
logger.error('Forcing shutdown...');
82
+
try {
83
+
logger.info('Shutting down gracefully...');
84
+
fs.writeFileSync('cursor.txt', jetstream.cursor!.toString(), 'utf8');
85
+
jetstream.close();
86
+
labelerServer.stop();
87
+
metricsServer.close();
88
+
} catch (error) {
89
+
logger.error(`Error shutting down gracefully: ${error}`);
79
90
process.exit(1);
80
-
}, 60000);
81
-
82
-
logger.info('Shutting down gracefully...');
83
-
jetstream.close();
84
-
labelerServer.stop();
85
-
metricsServer.close();
86
-
clearInterval(cursorUpdateInterval);
87
-
fs.writeFileSync('cursor.txt', cursor.toString(), 'utf8');
88
-
process.exit(0);
91
+
}
89
92
}
90
93
91
94
process.on('SIGINT', shutdown);
-32
src/types.ts
-32
src/types.ts
···
1
-
export interface EventStream {
2
-
did: string;
3
-
time_us: number;
4
-
type: string;
5
-
commit?: {
6
-
rev: string;
7
-
type: string;
8
-
collection: string;
9
-
rkey: string;
10
-
record: {
11
-
$type: string;
12
-
createdAt: string;
13
-
subject: {
14
-
cid: string;
15
-
uri: string;
16
-
};
17
-
};
18
-
};
19
-
}
20
-
21
-
export interface Label {
22
-
ver?: number;
23
-
src: string;
24
-
uri: string;
25
-
cid?: string;
26
-
val: string;
27
-
neg?: boolean;
28
-
cts: string;
29
-
exp?: string;
30
-
sig?: Uint8Array;
31
-
[k: string]: unknown;
32
-
}