[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
feat: add Dockerfile and ingest entrypoint for containerized Jetstream ingestion
- bin/ingest.ts: runtime-configurable entrypoint supporting postgres and sqlite adapters
- Dockerfile: Node 22 alpine image with pnpm, runs via tsx
- Config, database URL, and Jetstream URL provided via env vars and volume mounts
Tom Scanlan
(Mar 31, 2026, 9:50 AM EDT)
6890c4e2
7bd02388
+108
+14
Dockerfile
+14
Dockerfile
···
1
+
FROM node:22-alpine
2
+
3
+
RUN corepack enable && corepack prepare pnpm@latest --activate
4
+
5
+
WORKDIR /app
6
+
7
+
COPY package.json pnpm-lock.yaml pnpm-workspace.yaml ./
8
+
RUN pnpm install --frozen-lockfile
9
+
10
+
COPY src/ src/
11
+
COPY app/ app/
12
+
COPY bin/ bin/
13
+
14
+
CMD ["npx", "tsx", "bin/ingest.ts"]
+94
bin/ingest.ts
+94
bin/ingest.ts
···
1
+
/**
2
+
* Persistent Jetstream ingestion with runtime configuration.
3
+
*
4
+
* Connects to an externally-provided database and subscribes to Jetstream.
5
+
* The container does not own the database — it receives a connection URL.
6
+
*
7
+
* Environment:
8
+
* JETSTREAM_URL - Jetstream WebSocket URL (required)
9
+
* CONTRAIL_ADAPTER - "postgres" or "sqlite" (default: "postgres")
10
+
* DATABASE_URL - PostgreSQL connection string (required when adapter=postgres)
11
+
* SQLITE_PATH - SQLite file path (required when adapter=sqlite)
12
+
* CONTRAIL_CONFIG - Path to config file (default: "/app/config/contrail.config.ts")
13
+
* FLUSH_INTERVAL_MS - Batch flush interval in ms (default: 500)
14
+
*/
15
+
16
+
import { Contrail } from "../src/index.ts";
17
+
import type { ContrailConfig, Database } from "../src/index.ts";
18
+
19
+
const ADAPTER = process.env.CONTRAIL_ADAPTER ?? "postgres";
20
+
const JETSTREAM_URL = process.env.JETSTREAM_URL;
21
+
const CONFIG_PATH = process.env.CONTRAIL_CONFIG ?? "/app/config/contrail.config.ts";
22
+
const FLUSH_INTERVAL_MS = parseInt(process.env.FLUSH_INTERVAL_MS ?? "500", 10);
23
+
24
+
async function createDatabase(): Promise<{ db: Database; cleanup: () => Promise<void> }> {
25
+
if (ADAPTER === "postgres") {
26
+
const url = process.env.DATABASE_URL;
27
+
if (!url) {
28
+
console.error("DATABASE_URL is required when CONTRAIL_ADAPTER=postgres");
29
+
process.exit(1);
30
+
}
31
+
const pg = await import("pg");
32
+
const { createPostgresDatabase } = await import("../src/adapters/postgres.ts");
33
+
const pool = new pg.default.Pool({ connectionString: url });
34
+
return { db: createPostgresDatabase(pool), cleanup: () => pool.end() };
35
+
}
36
+
37
+
if (ADAPTER === "sqlite") {
38
+
const path = process.env.SQLITE_PATH;
39
+
if (!path) {
40
+
console.error("SQLITE_PATH is required when CONTRAIL_ADAPTER=sqlite");
41
+
process.exit(1);
42
+
}
43
+
const { createSqliteDatabase } = await import("../src/adapters/sqlite.ts");
44
+
return { db: createSqliteDatabase(path), cleanup: async () => {} };
45
+
}
46
+
47
+
console.error(`Unknown adapter: ${ADAPTER}. Use "postgres" or "sqlite".`);
48
+
process.exit(1);
49
+
}
50
+
51
+
async function loadConfig(): Promise<ContrailConfig> {
52
+
const module = await import(CONFIG_PATH);
53
+
return module.config ?? module.default;
54
+
}
55
+
56
+
async function main() {
57
+
if (!JETSTREAM_URL) {
58
+
console.error("JETSTREAM_URL is required");
59
+
process.exit(1);
60
+
}
61
+
62
+
const config = await loadConfig();
63
+
config.jetstreams = [JETSTREAM_URL];
64
+
65
+
const { db, cleanup } = await createDatabase();
66
+
const contrail = new Contrail({ ...config, db });
67
+
68
+
const controller = new AbortController();
69
+
const shutdown = () => {
70
+
console.log("\nShutting down gracefully...");
71
+
controller.abort();
72
+
};
73
+
process.on("SIGTERM", shutdown);
74
+
process.on("SIGINT", shutdown);
75
+
76
+
console.log("Contrail ingester starting");
77
+
console.log(` Adapter: ${ADAPTER}`);
78
+
console.log(` Jetstream: ${JETSTREAM_URL}`);
79
+
console.log(` Config: ${CONFIG_PATH}`);
80
+
console.log(` Flush: ${FLUSH_INTERVAL_MS}ms`);
81
+
82
+
await contrail.runPersistent({
83
+
signal: controller.signal,
84
+
flushIntervalMs: FLUSH_INTERVAL_MS,
85
+
});
86
+
87
+
await cleanup();
88
+
console.log("Done.");
89
+
}
90
+
91
+
main().catch((err) => {
92
+
console.error(err);
93
+
process.exit(1);
94
+
});