[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
some fixes
+101
-31
+3
-1
tests/notify.test.ts
+3
-1
tests/notify.test.ts
···
6
6
import { applyEvents, queryRecords } from "../src/core/db/records";
7
7
import type { Hono } from "hono";
8
8
9
+
const NOTIFY_CONFIG = { ...TEST_CONFIG, notify: true };
10
+
9
11
let db: Database;
10
12
let app: Hono;
11
13
12
14
beforeEach(async () => {
13
15
db = await createTestDbWithSchema();
14
-
app = createApp(db, TEST_CONFIG);
16
+
app = createApp(db, NOTIFY_CONFIG);
15
17
});
16
18
17
19
afterEach(() => {
+49
-7
src/core/client.ts
+49
-7
src/core/client.ts
···
18
18
pds: string | null;
19
19
}
20
20
21
+
/** Reject PDS URLs that point to private/internal addresses or non-HTTPS */
22
+
function validatePdsUrl(url: string): boolean {
23
+
try {
24
+
const parsed = new URL(url);
25
+
if (parsed.protocol !== "https:") return false;
26
+
const host = parsed.hostname;
27
+
// Block private/internal IP ranges
28
+
if (host === "localhost" || host === "127.0.0.1" || host === "[::1]") return false;
29
+
if (host.startsWith("10.")) return false;
30
+
if (host.startsWith("192.168.")) return false;
31
+
if (host.startsWith("169.254.")) return false;
32
+
if (/^172\.(1[6-9]|2\d|3[01])\./.test(host)) return false;
33
+
return true;
34
+
} catch {
35
+
return false;
36
+
}
37
+
}
38
+
21
39
async function resolveViaSlingshot(
22
40
identifier: string
23
41
): Promise<ResolvedIdentity | undefined> {
···
65
83
identifier: string
66
84
): Promise<ResolvedIdentity | undefined> {
67
85
const result = await resolveViaSlingshot(identifier);
68
-
if (result?.pds) return result;
86
+
if (result?.pds) {
87
+
if (!validatePdsUrl(result.pds)) return { ...result, pds: null };
88
+
return result;
89
+
}
69
90
70
91
// Fall back to DID doc resolution (only works for DIDs, not handles)
71
92
if (identifier.startsWith("did:")) {
72
93
try {
73
94
const pds = await getPDSViaDidDoc(identifier as Did);
74
-
if (pds) {
95
+
if (pds && validatePdsUrl(pds)) {
75
96
return {
76
97
did: identifier,
77
98
handle: result?.handle ?? null,
···
86
107
return result;
87
108
}
88
109
89
-
// In-memory PDS cache + in-flight deduplication
90
-
const pdsCache = new Map<string, string>();
110
+
// In-memory PDS cache with TTL + size limit, plus in-flight deduplication
111
+
const PDS_CACHE_TTL = 60 * 60 * 1000; // 1 hour
112
+
const PDS_CACHE_MAX = 10_000;
113
+
const pdsCache = new Map<string, { pds: string; at: number }>();
91
114
const pdsInflight = new Map<string, Promise<string | undefined>>();
115
+
116
+
function pdsCacheGet(did: string): string | undefined {
117
+
const entry = pdsCache.get(did);
118
+
if (!entry) return undefined;
119
+
if (Date.now() - entry.at > PDS_CACHE_TTL) {
120
+
pdsCache.delete(did);
121
+
return undefined;
122
+
}
123
+
return entry.pds;
124
+
}
125
+
126
+
function pdsCacheSet(did: string, pds: string): void {
127
+
// Evict oldest entries if over limit
128
+
if (pdsCache.size >= PDS_CACHE_MAX) {
129
+
const first = pdsCache.keys().next().value;
130
+
if (first) pdsCache.delete(first);
131
+
}
132
+
pdsCache.set(did, { pds, at: Date.now() });
133
+
}
92
134
93
135
export async function getPDS(
94
136
did: Did,
95
137
db?: Database
96
138
): Promise<string | undefined> {
97
-
const mem = pdsCache.get(did);
139
+
const mem = pdsCacheGet(did);
98
140
if (mem) return mem;
99
141
100
142
// Deduplicate concurrent calls for the same DID
···
120
162
.bind(did)
121
163
.first<{ pds: string }>();
122
164
if (cached?.pds) {
123
-
pdsCache.set(did, cached.pds);
165
+
pdsCacheSet(did, cached.pds);
124
166
return cached.pds;
125
167
}
126
168
}
···
128
170
const resolved = await resolvePDS(did);
129
171
if (!resolved?.pds) return undefined;
130
172
131
-
pdsCache.set(did, resolved.pds);
173
+
pdsCacheSet(did, resolved.pds);
132
174
133
175
// Persist to DB for future runs
134
176
if (db) {
+28
-22
src/core/persistent.ts
+28
-22
src/core/persistent.ts
···
116
116
117
117
const buffer: IngestEvent[] = [];
118
118
let flushDue = false;
119
+
let flushing = false;
119
120
let flushTimer: ReturnType<typeof setTimeout> | null = null;
120
121
121
122
const resetFlushTimer = () => {
···
124
125
};
125
126
126
127
const flush = async () => {
127
-
if (buffer.length === 0) return;
128
+
if (buffer.length === 0 || flushing) return;
129
+
flushing = true;
128
130
const batch = buffer.splice(0);
129
131
flushDue = false;
130
132
resetFlushTimer();
131
133
132
-
await applyEvents(db, batch, config);
134
+
try {
135
+
await applyEvents(db, batch, config);
133
136
134
-
const lastTimeUs = Math.max(...batch.map((e) => e.time_us));
135
-
await saveCursor(db, lastTimeUs);
137
+
const lastTimeUs = Math.max(...batch.map((e) => e.time_us));
138
+
await saveCursor(db, lastTimeUs);
136
139
137
-
// Identity refresh
138
-
const uniqueDids = [...new Set(batch.map((e) => e.did))];
139
-
if (uniqueDids.length > 0) {
140
-
try {
141
-
await refreshStaleIdentities(db, uniqueDids);
142
-
} catch (err) {
143
-
log.warn(`Identity refresh failed: ${err}`);
140
+
// Identity refresh
141
+
const uniqueDids = [...new Set(batch.map((e) => e.did))];
142
+
if (uniqueDids.length > 0) {
143
+
try {
144
+
await refreshStaleIdentities(db, uniqueDids);
145
+
} catch (err) {
146
+
log.warn(`Identity refresh failed: ${err}`);
147
+
}
144
148
}
145
-
}
146
149
147
-
// Feed pruning
148
-
if (config.feeds && Date.now() - state.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) {
149
-
const maxItems = Math.max(
150
-
...Object.values(config.feeds).map((f) => f.maxItems ?? DEFAULT_FEED_MAX_ITEMS)
151
-
);
152
-
const pruned = await pruneFeedItems(db, maxItems);
153
-
if (pruned > 0) log.log(`Pruned ${pruned} old feed items`);
154
-
state.lastFeedPruneMs = Date.now();
155
-
}
150
+
// Feed pruning
151
+
if (config.feeds && Date.now() - state.lastFeedPruneMs > FEED_PRUNE_INTERVAL_MS) {
152
+
const maxItems = Math.max(
153
+
...Object.values(config.feeds).map((f) => f.maxItems ?? DEFAULT_FEED_MAX_ITEMS)
154
+
);
155
+
const pruned = await pruneFeedItems(db, maxItems);
156
+
if (pruned > 0) log.log(`Pruned ${pruned} old feed items`);
157
+
state.lastFeedPruneMs = Date.now();
158
+
}
156
159
157
-
log.log(`Flushed ${batch.length} events. Cursor: ${lastTimeUs}`);
160
+
log.log(`Flushed ${batch.length} events. Cursor: ${lastTimeUs}`);
161
+
} finally {
162
+
flushing = false;
163
+
}
158
164
};
159
165
160
166
// Handle abort
+4
src/core/types.ts
+4
src/core/types.ts
···
106
106
jetstreams?: string[];
107
107
feeds?: Record<string, FeedConfig>;
108
108
logger?: Logger;
109
+
/** Expose the notifyOfUpdate HTTP endpoint. Off by default.
110
+
* Set to `true` for open access, or a string to require `Authorization: Bearer <secret>`. */
111
+
notify?: boolean | string;
109
112
}
110
113
111
114
export interface ResolvedRelation {
···
237
240
for (const [, rel] of Object.entries(colConfig.relations ?? {})) {
238
241
if (rel.field) validateFieldName(rel.field);
239
242
if (rel.groupBy) validateFieldName(rel.groupBy);
243
+
if (rel.countDistinct) validateFieldName(rel.countDistinct);
240
244
}
241
245
if (Array.isArray(colConfig.searchable)) {
242
246
for (const field of colConfig.searchable) {
+1
-1
src/core/router/hydrate.ts
+1
-1
src/core/router/hydrate.ts
+11
src/core/router/notify.ts
+11
src/core/router/notify.ts
···
141
141
db: Database,
142
142
config: ContrailConfig
143
143
) {
144
+
// Endpoint is off by default. Set config.notify to true or a secret string to enable.
145
+
if (!config.notify) return;
146
+
144
147
const ns = config.namespace;
148
+
const secret = typeof config.notify === "string" ? config.notify : null;
145
149
146
150
app.post(`/xrpc/${ns}.notifyOfUpdate`, async (c) => {
151
+
if (secret) {
152
+
const auth = c.req.header("Authorization");
153
+
if (auth !== `Bearer ${secret}`) {
154
+
return c.json({ error: "unauthorized" }, 401);
155
+
}
156
+
}
157
+
147
158
const body = await c.req.json<{ uri?: string; uris?: string[] }>().catch(() => null);
148
159
const uris: string[] = [];
149
160