[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
feat(contrail): private-network deployment + concurrent-init hardening (#44)
* feat(contrail-base): networkOverrides config for private-network deployments
Add optional ContrailConfig.networkOverrides with three subfields:
plcUrl, slingshotUrl, additionalAllowedHosts. Plumb config through
resolvePDS, getPDSViaDidDoc (inlined; module-level didResolver removed),
getPDS, resolvePDSCached, getClient, and all five identity.ts functions
(fetchAndSave, resolveIdentity, resolveIdentities, resolveActor,
refreshStaleIdentities). Pass config through backfill.ts call sites
(getClient + getPDS) and the persistent.ts:160 refreshStaleIdentities
call. additionalAllowedHosts is exact + case-insensitive + port-agnostic;
the default SSRF validator still runs for non-listed hosts.
All changes are additive — omitting networkOverrides preserves current
public-internet defaults. validatePdsUrl remains internal (no new
exports beyond the networkOverrides field itself).
* test(contrail-base): network overrides + SSRF regression coverage
Add three test files exercising the networkOverrides plumbing:
- client.test.ts: ContrailConfig.networkOverrides type, validatePdsUrl
regression baseline (public HTTPS accepted; non-HTTPS + private CIDRs
+ localhost + link-local rejected), additionalAllowedHosts allowlist
(allowed host accepted; non-listed hosts still rejected; port-agnostic
match), getClient + getPDS + resolvePDS plumb-through with mocked fetch.
- identity-config.test.ts: resolveIdentity, resolveIdentities, resolveActor,
refreshStaleIdentities all route to override slingshot URL.
- network-overrides.test.ts: integration test against node:http stub PLC
and Slingshot servers. Asserts resolvePDS, getClient, and
refreshStaleIdentities actually hit the configured override URLs end to
end, exercising the full chain.
The SSRF regression cases give the existing validator implicit baseline
explicit coverage in this PR.
* feat(contrail-base): apply networkOverrides.resolver to spaces credential verifier
* feat(contrail-appview): apply networkOverrides to label-endpoint resolution
* fix(contrail-appview): use IF NOT EXISTS on schema migrations to remove init race
Concurrent `initSchema` calls previously raced on `ALTER TABLE ADD COLUMN`
because the SQL had no `IF NOT EXISTS` clause, and the two consumer-site
`try { ... } catch { /* Column already exists — ignore */ }` blocks silently
swallowed *every* DDL error — masking missing tables, syntax errors, and
type mismatches alongside the intended duplicate-column case.
L3 replaces both swallows with a dialect-aware `addColumnIfNotExists` helper:
- Postgres: emits `ALTER TABLE ... ADD COLUMN IF NOT EXISTS` (atomic in PG).
- SQLite: pre-checks `PRAGMA table_info`, then narrowly absorbs only the
literal "duplicate column name" error from the still-possible PRAGMA->ALTER
race. Every other error surfaces.
The `MIGRATIONS` array is now structured `{ table, column, columnDef, target }`
data routed through the same helper. `target: "spaces"` routes to `spacesDb`
when one is configured (split-DB deployments), `target: "feeds"` is gated on
feeds being configured so the absent `feed_backfills` table no longer
silently fails.
Tests at `packages/contrail/tests/schema-idempotency.test.ts` cover:
- sequential and concurrent (3-way) `initSchema` calls
- repeated init does not leave duplicate count columns
- helper is a no-op when column already exists
- helper surfaces "no such table" errors (the L3-removed swallow case)
- `initSchema` propagates errors thrown from extension schema modules
OpenMeet's consumer-side Postgres advisory lock around `contrail.init()`
exists only to work around this race; with L3 in place that lock can be
deleted in a follow-up.
* test: L3 concurrent CREATE race on Postgres
* chore: add changeset for private-network deployment support
* fix(contrail-appview): thread networkOverrides config through all resolver/identity call sites
The networkOverrides config param was threaded into the resolver/identity
functions but dropped at ~8 appview call sites (live-ingest refresh cycle,
on-demand refresh, and router getProfile/getFeed/collection/profiles/notify),
so private-network deploys silently fell back to the public resolver and the
un-widened SSRF guard on those paths.
Pass the in-scope config at each site. Add a regression test driving an
override config through the live-ingest refresh path (runIngestCycle) and a
router actor-resolution path (getProfile), asserting the override slingshot
is hit instead of the default.
* refactor(contrail-base): dedup SSRF validator into shared validateExternalUrl
labels/resolve.ts validateEndpointUrl duplicated contrail-base validatePdsUrl,
and this PR had edited the additionalAllowedHosts allowlist into both copies —
a future SSRF-rule fix applied to one copy would leave the other exploitable.
Export a single validateExternalUrl(url, allowedHosts?) from contrail-base and
consume it from both the PDS client and labeler-endpoint resolution.
validateEndpointUrl stays exported as a thin alias for backward compat.
Behavior is identical (covered by existing SSRF tests).
Also document the PDS resolve cache's process-wide-config assumption (keyed by
DID only; safe now that all callers thread the same config).
* docs(contrail-base): document validateExternalUrl threat-model scope
Note inline that the SSRF guard is best-effort literal-match only: no DNS
resolution, partial IPv6 / encoded-IP coverage, egress network policy
expected for fully untrusted resolver inputs. No behavior change.
authored by
Tom Scanlan
and committed by
GitHub
(Jun 5, 2026, 10:21 AM EDT)
d7e09366
1fcca396
+1449
-129
+19
.changeset/private-network-overrides.md
+19
.changeset/private-network-overrides.md
···
1
+
---
2
+
"@atmo-dev/contrail-base": minor
3
+
"@atmo-dev/contrail-appview": minor
4
+
"@atmo-dev/contrail-community": minor
5
+
---
6
+
7
+
Private-network deployment support via a new optional `ContrailConfig.networkOverrides` block.
8
+
9
+
`networkOverrides` carries three optional subfields, all defaulting to the current public-internet behavior (omit the block entirely and nothing changes):
10
+
11
+
- **`resolver`** — a custom `DidDocumentResolver` used during DID-doc PDS fallback, labeler-endpoint resolution, and spaces service-auth JWT verification. Lets a deployment point at a private PLC mirror or inject a custom fetch (mTLS, retry, instrumentation). Trusted; not SSRF-checked.
12
+
- **`slingshotUrl`** — override the slingshot identity-resolver endpoint. Trusted; not SSRF-checked.
13
+
- **`additionalAllowedHosts`** — hostnames that bypass the default SSRF guard when validating a resolved PDS or labeler endpoint. Match is exact, case-insensitive, port-agnostic (e.g. `["pds.dev.svc.cluster.local"]`). This is the only knob that widens the validator; there is no "disable SSRF" flag.
14
+
15
+
The overrides are threaded through PDS/identity resolution (`resolvePDS`, `getPDS`, `getClient`, `resolveIdentity*`, `refreshStaleIdentities`), labeler endpoint resolution and ingest (`resolveLabelerEndpoint`, `getLabelerState`, label subscribe cycles), and service-auth verification (`buildVerifier` in both the appview router and the community integration). The in-scope `config` is now also passed at every appview call site that resolves identities or PDS endpoints — the live-ingest refresh cycle (`runIngestCycle` → `refreshStaleIdentities`), the on-demand `refresh` path, and the router actor/identity/PDS resolution paths (`getProfile`, `getFeed`, collection queries, profile hydration, notify) — so private-network deploys honor the override on those paths instead of silently falling back to the public resolver and un-widened SSRF guard.
16
+
17
+
The SSRF guard is now a single shared validator: `validateExternalUrl(url, additionalAllowedHosts?)` is exported from `contrail-base` and consumed by both the PDS client and labeler-endpoint resolution. `validateEndpointUrl` remains exported as a thin alias for backward compatibility. This removes the previous duplicate validator (`validatePdsUrl` + `validateEndpointUrl`) where an allowlist or SSRF-rule edit could be applied to only one copy.
18
+
19
+
Also hardens schema initialization for concurrent/Postgres deployments: a dialect-aware `addColumnIfNotExists` (Postgres `ADD COLUMN IF NOT EXISTS`; SQLite pre-check), narrow absorption of the Postgres concurrent-`CREATE` race (42P07 / 23505 on pg_type/pg_class/pg_namespace indexes), and per-statement (rather than batched) DDL during `initSchema` / `initSpacesSchema` / spaces schema. Genuine DDL errors (syntax, type mismatch, missing column/table) still propagate.
+81
-30
packages/contrail-base/src/client.ts
+81
-30
packages/contrail-base/src/client.ts
···
2
2
CompositeDidDocumentResolver,
3
3
PlcDidDocumentResolver,
4
4
WebDidDocumentResolver,
5
+
type DidDocumentResolver,
5
6
} from "@atcute/identity-resolver";
6
7
import { type Did } from "@atcute/lexicons";
7
8
import { Client, simpleFetchHandler } from "@atcute/client";
8
9
import type {} from "@atcute/atproto";
9
-
import type { Database } from "./types";
10
+
import type { ContrailConfig, Database } from "./types";
10
11
11
12
// Slingshot-first PDS resolution with fallback to DID document resolution
12
13
const SLINGSHOT_URL =
···
18
19
pds: string | null;
19
20
}
20
21
21
-
/** Reject PDS URLs that point to private/internal addresses or non-HTTPS */
22
-
function validatePdsUrl(url: string): boolean {
22
+
/** Reject external URLs (PDS, labeler, …) that point to private/internal
23
+
* addresses or non-HTTPS. The single SSRF guard shared across packages —
24
+
* callers MUST route every externally-resolved endpoint through this so the
25
+
* allowlist rules live in exactly one place.
26
+
*
27
+
* Hostnames in `additionalAllowedHosts` skip both checks. Match is exact,
28
+
* case-insensitive (allowlist entries are lowercased on compare; `URL.hostname`
29
+
* is already lowercased), and port-agnostic.
30
+
*
31
+
* Scope: best-effort guard against the obvious internal-address classes
32
+
* (private/link-local IPv4 literals, localhost, non-HTTPS). It does NOT
33
+
* resolve DNS, so a public hostname that resolves to a private address is not
34
+
* caught here, and IPv6 / non-canonical IP encodings are only partially
35
+
* covered. Defense-in-depth (egress network policy) is expected when resolver
36
+
* inputs are fully untrusted. */
37
+
export function validateExternalUrl(url: string, additionalAllowedHosts?: string[]): boolean {
38
+
let parsed: URL;
23
39
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;
40
+
parsed = new URL(url);
34
41
} catch {
35
42
return false;
36
43
}
44
+
if (additionalAllowedHosts?.some((h) => h.toLowerCase() === parsed.hostname)) {
45
+
return true;
46
+
}
47
+
if (parsed.protocol !== "https:") return false;
48
+
const host = parsed.hostname;
49
+
// Block private/internal IP ranges
50
+
if (host === "localhost" || host === "127.0.0.1" || host === "[::1]") return false;
51
+
if (host.startsWith("10.")) return false;
52
+
if (host.startsWith("192.168.")) return false;
53
+
if (host.startsWith("169.254.")) return false;
54
+
if (/^172\.(1[6-9]|2\d|3[01])\./.test(host)) return false;
55
+
return true;
37
56
}
38
57
39
58
async function resolveViaSlingshot(
40
-
identifier: string
59
+
identifier: string,
60
+
slingshotUrl: string,
41
61
): Promise<ResolvedIdentity | undefined> {
42
-
const url = new URL(SLINGSHOT_URL);
62
+
const url = new URL(slingshotUrl);
43
63
url.searchParams.set("identifier", identifier);
44
64
45
65
try {
···
61
81
}
62
82
}
63
83
64
-
const didResolver = new CompositeDidDocumentResolver({
84
+
const DEFAULT_DID_RESOLVER: DidDocumentResolver = new CompositeDidDocumentResolver({
65
85
methods: {
66
86
plc: new PlcDidDocumentResolver(),
67
87
web: new WebDidDocumentResolver(),
68
88
},
69
89
});
70
90
71
-
async function getPDSViaDidDoc(did: Did): Promise<string | undefined> {
72
-
const doc = await didResolver.resolve(did as Did<"plc"> | Did<"web">);
91
+
async function getPDSViaDidDoc(
92
+
did: Did,
93
+
config?: ContrailConfig,
94
+
): Promise<string | undefined> {
95
+
const resolver = config?.networkOverrides?.resolver ?? DEFAULT_DID_RESOLVER;
96
+
const doc = await resolver.resolve(did as Did<"plc"> | Did<"web">);
73
97
return doc.service
74
98
?.find((s) => s.id === "#atproto_pds")
75
99
?.serviceEndpoint.toString();
···
78
102
/**
79
103
* Resolve identity info (did, handle, pds) for a DID or handle.
80
104
* Uses slingshot first, falls back to DID doc for PDS.
105
+
*
106
+
* `config?.networkOverrides` (optional): customize the slingshot endpoint,
107
+
* the PLC URL used during DID-doc fallback, and/or which hostnames bypass
108
+
* the default SSRF guard. Omitting `config` preserves all defaults.
81
109
*/
82
110
export async function resolvePDS(
83
-
identifier: string
111
+
identifier: string,
112
+
config?: ContrailConfig,
84
113
): Promise<ResolvedIdentity | undefined> {
85
-
const result = await resolveViaSlingshot(identifier);
114
+
const slingshotUrl = config?.networkOverrides?.slingshotUrl ?? SLINGSHOT_URL;
115
+
const allowed = config?.networkOverrides?.additionalAllowedHosts;
116
+
const result = await resolveViaSlingshot(identifier, slingshotUrl);
86
117
if (result?.pds) {
87
-
if (!validatePdsUrl(result.pds)) return { ...result, pds: null };
118
+
if (!validateExternalUrl(result.pds, allowed)) return { ...result, pds: null };
88
119
return result;
89
120
}
90
121
91
122
// Fall back to DID doc resolution (only works for DIDs, not handles)
92
123
if (identifier.startsWith("did:")) {
93
124
try {
94
-
const pds = await getPDSViaDidDoc(identifier as Did);
95
-
if (pds && validatePdsUrl(pds)) {
125
+
const pds = await getPDSViaDidDoc(identifier as Did, config);
126
+
if (pds && validateExternalUrl(pds, allowed)) {
96
127
return {
97
128
did: identifier,
98
129
handle: result?.handle ?? null,
···
107
138
return result;
108
139
}
109
140
110
-
// In-memory PDS cache with TTL + size limit, plus in-flight deduplication
141
+
// In-memory PDS cache with TTL + size limit, plus in-flight deduplication.
142
+
//
143
+
// Keyed by DID only — this assumes a single, process-wide `networkOverrides`
144
+
// config (the deployment model: one resolver + one SSRF allowlist per process).
145
+
// Every caller in this monorepo now threads the same in-scope `config`, so a
146
+
// config-less and an override-aware resolution can never race for the same DID.
147
+
// If a future deployment ever resolves the same DID under differing overrides
148
+
// in one process, key these caches by an override fingerprint instead.
111
149
const PDS_CACHE_TTL = 60 * 60 * 1000; // 1 hour
112
150
const PDS_CACHE_MAX = 10_000;
113
151
const pdsCache = new Map<string, { pds: string; at: number }>();
···
134
172
135
173
export async function getPDS(
136
174
did: Did,
137
-
db?: Database
175
+
db?: Database,
176
+
config?: ContrailConfig,
138
177
): Promise<string | undefined> {
139
178
const mem = pdsCacheGet(did);
140
179
if (mem) return mem;
···
143
182
const inflight = pdsInflight.get(did);
144
183
if (inflight) return inflight;
145
184
146
-
const promise = resolvePDSCached(did, db);
185
+
const promise = resolvePDSCached(did, db, config);
147
186
pdsInflight.set(did, promise);
148
187
try {
149
188
return await promise;
···
154
193
155
194
async function resolvePDSCached(
156
195
did: Did,
157
-
db?: Database
196
+
db?: Database,
197
+
config?: ContrailConfig,
158
198
): Promise<string | undefined> {
159
199
if (db) {
160
200
const cached = await db
···
167
207
}
168
208
}
169
209
170
-
const resolved = await resolvePDS(did);
210
+
const resolved = await resolvePDS(did, config);
171
211
if (!resolved?.pds) return undefined;
172
212
173
213
pdsCacheSet(did, resolved.pds);
···
185
225
return resolved.pds;
186
226
}
187
227
188
-
export async function getClient(did: Did, db?: Database): Promise<Client> {
189
-
const pds = await getPDS(did, db);
228
+
export async function getClient(
229
+
did: Did,
230
+
db?: Database,
231
+
config?: ContrailConfig,
232
+
): Promise<Client> {
233
+
const pds = await getPDS(did, db, config);
190
234
if (!pds) throw new Error(`PDS not found for ${did}`);
191
235
return new Client({
192
236
handler: simpleFetchHandler({ service: pds }),
193
237
});
238
+
}
239
+
240
+
/** Test-only: clear module-level PDS caches. Production code MUST NOT call this.
241
+
* Exported with a `__` prefix to signal it is not part of the public API. */
242
+
export function __resetPdsCachesForTests(): void {
243
+
pdsCache.clear();
244
+
pdsInflight.clear();
194
245
}
+16
-11
packages/contrail-base/src/identity.ts
+16
-11
packages/contrail-base/src/identity.ts
···
1
1
import type { Did } from "@atcute/lexicons";
2
-
import type { Database, Logger } from "./types";
2
+
import type { ContrailConfig, Database, Logger } from "./types";
3
3
import { isDid, isHandle } from "@atcute/lexicons/syntax";
4
4
import { resolvePDS } from "./client";
5
5
···
28
28
async function fetchAndSave(
29
29
db: Database,
30
30
identifier: string,
31
-
cached?: Identity | null
31
+
cached?: Identity | null,
32
+
config?: ContrailConfig,
32
33
): Promise<Identity> {
33
-
const resolved = await resolvePDS(identifier);
34
+
const resolved = await resolvePDS(identifier, config);
34
35
const identity: Identity = {
35
36
did: resolved?.did ?? identifier,
36
37
handle: resolved?.handle ?? cached?.handle ?? null,
···
43
44
44
45
export async function resolveIdentity(
45
46
db: Database,
46
-
did: Did
47
+
did: Did,
48
+
config?: ContrailConfig,
47
49
): Promise<Identity> {
48
50
const cached = await db
49
51
.prepare("SELECT did, handle, pds, resolved_at FROM identities WHERE did = ?")
···
52
54
53
55
if (cached && !isStale(cached.resolved_at)) return cached;
54
56
55
-
return fetchAndSave(db, did, cached);
57
+
return fetchAndSave(db, did, cached, config);
56
58
}
57
59
58
60
export async function resolveIdentities(
59
61
db: Database,
60
-
dids: string[]
62
+
dids: string[],
63
+
config?: ContrailConfig,
61
64
): Promise<Map<string, Identity>> {
62
65
const map = new Map<string, Identity>();
63
66
if (dids.length === 0) return map;
···
80
83
for (const did of dids) {
81
84
if (map.has(did) || !isDid(did)) continue;
82
85
try {
83
-
const identity = await fetchAndSave(db, did);
86
+
const identity = await fetchAndSave(db, did, undefined, config);
84
87
map.set(did, identity);
85
88
} catch {
86
89
// Silently skip unresolvable identities
···
92
95
93
96
export async function resolveActor(
94
97
db: Database,
95
-
actor: string
98
+
actor: string,
99
+
config?: ContrailConfig,
96
100
): Promise<string | null> {
97
101
if (isDid(actor)) return actor;
98
102
if (!isHandle(actor)) return null;
···
106
110
if (cached && !isStale(cached.resolved_at)) return cached.did;
107
111
108
112
// Resolve via slingshot
109
-
const resolved = await resolvePDS(actor);
113
+
const resolved = await resolvePDS(actor, config);
110
114
if (!resolved?.did || !isDid(resolved.did)) return null;
111
115
112
116
await saveIdentity(db, {
···
139
143
140
144
export async function refreshStaleIdentities(
141
145
db: Database,
142
-
dids: string[]
146
+
dids: string[],
147
+
config?: ContrailConfig,
143
148
): Promise<void> {
144
149
if (dids.length === 0) return;
145
150
···
169
174
170
175
for (const did of toRefresh) {
171
176
try {
172
-
await fetchAndSave(db, did);
177
+
await fetchAndSave(db, did, undefined, config);
173
178
} catch {
174
179
// Silently skip unresolvable identities
175
180
}
+31
packages/contrail-base/src/types.ts
+31
packages/contrail-base/src/types.ts
···
241
241
* ingests synthesized rows for any follower already in our identities
242
242
* table. Lets newcomers immediately appear in existing users' feeds. */
243
243
constellation?: ConstellationConfig | false;
244
+
/** Network overrides for private-network or test deployments.
245
+
* All subfields default to current public-internet behavior;
246
+
* omitting `networkOverrides` entirely preserves current behavior.
247
+
*
248
+
* SECURITY: `resolver` and `slingshotUrl` are taken at face value and are
249
+
* NOT validated against the SSRF guard — the consumer is trusted to
250
+
* configure them. Only the PDS URL returned downstream is validated,
251
+
* and only `additionalAllowedHosts` widens that PDS validator. There is
252
+
* no "disable SSRF" flag. */
253
+
networkOverrides?: {
254
+
/** DID document resolver used during the DID-doc PDS fallback. When
255
+
* unset, contrail constructs a default `CompositeDidDocumentResolver`
256
+
* with PLC + Web methods pointing at the upstream PLC directory.
257
+
* Pass a custom resolver to point at a private PLC mirror, inject a
258
+
* custom fetch (mTLS, retry, instrumentation), or swap in an
259
+
* alternative DID method composition.
260
+
* Mirrors the `AuthorityConfig.resolver` pattern in `spaces/types.ts`. */
261
+
resolver?: import("@atcute/identity-resolver").DidDocumentResolver;
262
+
/** Slingshot identity resolver URL override. Trusted; not SSRF-checked.
263
+
* Default: https://slingshot.microcosm.blue/xrpc/com.bad-example.identity.resolveMiniDoc */
264
+
slingshotUrl?: string;
265
+
/** Hostnames (DNS names or IP literals) to allow past the default SSRF
266
+
* guard when validating a resolved PDS URL.
267
+
* For listed hostnames, the non-HTTPS + private-CIDR checks are skipped.
268
+
* For all other hostnames, the default validator runs unchanged.
269
+
* Match semantics: exact hostname, case-insensitive (entries are
270
+
* lowercased on comparison; `URL.hostname` is already lowercased),
271
+
* port-agnostic.
272
+
* Example: ["pds.dev.svc.cluster.local"]. */
273
+
additionalAllowedHosts?: string[];
274
+
};
244
275
}
245
276
246
277
export interface ConstellationConfig {
+1
-1
packages/contrail-community/src/integration.ts
+1
-1
packages/contrail-community/src/integration.ts
···
60
60
registerRoutes(app, opts) {
61
61
// Reuse the spaces JWT verifier — the auth model is identical.
62
62
if (!config.spaces?.authority) return;
63
-
const verifier = buildVerifier(config.spaces.authority);
63
+
const verifier = buildVerifier(config.spaces.authority, config.networkOverrides);
64
64
const authMiddleware =
65
65
opts?.authMiddleware ?? createServiceAuthMiddleware(verifier);
66
66
registerCommunityRoutes(
+239
packages/contrail/tests/client.test.ts
+239
packages/contrail/tests/client.test.ts
···
1
+
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
2
+
import { resolvePDS, getClient, getPDS, __resetPdsCachesForTests } from "../src/core/client";
3
+
import { type DidDocumentResolver } from "@atcute/identity-resolver";
4
+
import { createTestDbWithSchema } from "./helpers";
5
+
import type { Did } from "@atcute/lexicons";
6
+
7
+
describe("validatePdsUrl via resolvePDS — regression baseline", () => {
8
+
let fetchSpy: ReturnType<typeof vi.spyOn>;
9
+
10
+
beforeEach(() => {
11
+
fetchSpy = vi.spyOn(global, "fetch");
12
+
});
13
+
14
+
afterEach(() => {
15
+
fetchSpy.mockRestore();
16
+
});
17
+
18
+
it("accepts a public HTTPS PDS", async () => {
19
+
fetchSpy.mockResolvedValue(
20
+
new Response(JSON.stringify({ did: "did:plc:a", handle: "a.bsky.social", pds: "https://shimeji.us-east.host.bsky.network" }), { status: 200 })
21
+
);
22
+
const r = await resolvePDS("did:plc:a");
23
+
expect(r?.pds).toBe("https://shimeji.us-east.host.bsky.network");
24
+
});
25
+
26
+
it("rejects non-HTTPS PDS (returns pds: null)", async () => {
27
+
fetchSpy.mockResolvedValue(
28
+
new Response(JSON.stringify({ did: "did:plc:b", handle: "b.test", pds: "http://malicious.example.com" }), { status: 200 })
29
+
);
30
+
const r = await resolvePDS("did:plc:b");
31
+
expect(r?.pds).toBe(null);
32
+
});
33
+
34
+
it("rejects 10.x private CIDR", async () => {
35
+
fetchSpy.mockResolvedValue(
36
+
new Response(JSON.stringify({ did: "did:plc:c", pds: "https://10.0.0.1" }), { status: 200 })
37
+
);
38
+
const r = await resolvePDS("did:plc:c");
39
+
expect(r?.pds).toBe(null);
40
+
});
41
+
42
+
it("rejects 192.168.x private CIDR", async () => {
43
+
fetchSpy.mockResolvedValue(
44
+
new Response(JSON.stringify({ did: "did:plc:d", pds: "https://192.168.1.1" }), { status: 200 })
45
+
);
46
+
const r = await resolvePDS("did:plc:d");
47
+
expect(r?.pds).toBe(null);
48
+
});
49
+
50
+
it("rejects 172.16-31.x private CIDR", async () => {
51
+
fetchSpy.mockResolvedValue(
52
+
new Response(JSON.stringify({ did: "did:plc:e", pds: "https://172.20.0.5" }), { status: 200 })
53
+
);
54
+
const r = await resolvePDS("did:plc:e");
55
+
expect(r?.pds).toBe(null);
56
+
});
57
+
58
+
it("rejects localhost and 169.254 link-local", async () => {
59
+
fetchSpy.mockResolvedValueOnce(
60
+
new Response(JSON.stringify({ did: "did:plc:f", pds: "https://localhost" }), { status: 200 })
61
+
);
62
+
expect((await resolvePDS("did:plc:f"))?.pds).toBe(null);
63
+
64
+
fetchSpy.mockResolvedValueOnce(
65
+
new Response(JSON.stringify({ did: "did:plc:g", pds: "https://169.254.169.254" }), { status: 200 })
66
+
);
67
+
expect((await resolvePDS("did:plc:g"))?.pds).toBe(null);
68
+
});
69
+
});
70
+
71
+
describe("validatePdsUrl via resolvePDS — additionalAllowedHosts allowlist", () => {
72
+
let fetchSpy: ReturnType<typeof vi.spyOn>;
73
+
74
+
beforeEach(() => {
75
+
fetchSpy = vi.spyOn(global, "fetch");
76
+
});
77
+
78
+
afterEach(() => {
79
+
fetchSpy.mockRestore();
80
+
});
81
+
82
+
it("accepts http://pds.dev.svc.cluster.local when host is on allowlist", async () => {
83
+
fetchSpy.mockResolvedValue(
84
+
new Response(JSON.stringify({ did: "did:plc:h", pds: "http://pds.dev.svc.cluster.local" }), { status: 200 })
85
+
);
86
+
const r = await resolvePDS("did:plc:h", {
87
+
namespace: "test",
88
+
collections: {},
89
+
networkOverrides: { additionalAllowedHosts: ["pds.dev.svc.cluster.local"] },
90
+
});
91
+
expect(r?.pds).toBe("http://pds.dev.svc.cluster.local");
92
+
});
93
+
94
+
it("still rejects http://other.private.host when not on allowlist", async () => {
95
+
fetchSpy.mockResolvedValue(
96
+
new Response(JSON.stringify({ did: "did:plc:i", pds: "http://other.private.host" }), { status: 200 })
97
+
);
98
+
const r = await resolvePDS("did:plc:i", {
99
+
namespace: "test",
100
+
collections: {},
101
+
networkOverrides: { additionalAllowedHosts: ["pds.dev.svc.cluster.local"] },
102
+
});
103
+
expect(r?.pds).toBe(null);
104
+
});
105
+
106
+
it("still rejects http://192.168.1.1 when not on allowlist", async () => {
107
+
fetchSpy.mockResolvedValue(
108
+
new Response(JSON.stringify({ did: "did:plc:j", pds: "http://192.168.1.1" }), { status: 200 })
109
+
);
110
+
const r = await resolvePDS("did:plc:j", {
111
+
namespace: "test",
112
+
collections: {},
113
+
networkOverrides: { additionalAllowedHosts: ["pds.dev.svc.cluster.local"] },
114
+
});
115
+
expect(r?.pds).toBe(null);
116
+
});
117
+
118
+
it("port-agnostic: allowlist matches hostname regardless of port", async () => {
119
+
fetchSpy.mockResolvedValue(
120
+
new Response(JSON.stringify({ did: "did:plc:k", pds: "http://pds.dev.svc.cluster.local:8080" }), { status: 200 })
121
+
);
122
+
const r = await resolvePDS("did:plc:k", {
123
+
namespace: "test",
124
+
collections: {},
125
+
networkOverrides: { additionalAllowedHosts: ["pds.dev.svc.cluster.local"] },
126
+
});
127
+
expect(r?.pds).toBe("http://pds.dev.svc.cluster.local:8080");
128
+
});
129
+
130
+
it("case-insensitive: mixed-case allowlist entries match lowercase URL.hostname", async () => {
131
+
fetchSpy.mockResolvedValue(
132
+
new Response(JSON.stringify({ did: "did:plc:l", pds: "http://pds.dev.svc.cluster.local" }), { status: 200 })
133
+
);
134
+
const r = await resolvePDS("did:plc:l", {
135
+
namespace: "test",
136
+
collections: {},
137
+
networkOverrides: { additionalAllowedHosts: ["PDS.Dev.Svc.Cluster.Local"] },
138
+
});
139
+
expect(r?.pds).toBe("http://pds.dev.svc.cluster.local");
140
+
});
141
+
});
142
+
143
+
describe("getPDSViaDidDoc — resolver override", () => {
144
+
let fetchSpy: ReturnType<typeof vi.spyOn>;
145
+
146
+
beforeEach(() => {
147
+
__resetPdsCachesForTests();
148
+
fetchSpy = vi.spyOn(global, "fetch");
149
+
});
150
+
151
+
afterEach(() => {
152
+
fetchSpy.mockRestore();
153
+
});
154
+
155
+
it("falls back to DID doc when slingshot returns no pds, and uses injected resolver", async () => {
156
+
fetchSpy.mockResolvedValueOnce(
157
+
new Response(JSON.stringify({ did: "did:plc:abc", handle: "alice.test" }), { status: 200 })
158
+
);
159
+
160
+
const resolveCalls: string[] = [];
161
+
const injectedResolver: DidDocumentResolver = {
162
+
async resolve(did: string) {
163
+
resolveCalls.push(did);
164
+
return {
165
+
service: [
166
+
{ id: "#atproto_pds", type: "AtprotoPersonalDataServer", serviceEndpoint: "http://pds.dev.svc.cluster.local" },
167
+
],
168
+
} as any;
169
+
},
170
+
} as any;
171
+
172
+
const r = await resolvePDS("did:plc:abc", {
173
+
namespace: "test",
174
+
collections: {},
175
+
networkOverrides: {
176
+
resolver: injectedResolver,
177
+
additionalAllowedHosts: ["pds.dev.svc.cluster.local"],
178
+
},
179
+
});
180
+
expect(r?.pds).toBe("http://pds.dev.svc.cluster.local");
181
+
expect(resolveCalls).toContain("did:plc:abc");
182
+
});
183
+
});
184
+
185
+
describe("getClient + getPDS — config plumb-through", () => {
186
+
let fetchSpy: ReturnType<typeof vi.spyOn>;
187
+
188
+
beforeEach(() => {
189
+
__resetPdsCachesForTests();
190
+
fetchSpy = vi.spyOn(global, "fetch");
191
+
});
192
+
193
+
afterEach(() => {
194
+
fetchSpy.mockRestore();
195
+
});
196
+
197
+
it("getClient with config?.networkOverrides.slingshotUrl uses the override", async () => {
198
+
fetchSpy.mockImplementation(async (input) => {
199
+
const url = String(input);
200
+
if (url.includes("my-slingshot")) {
201
+
return new Response(
202
+
JSON.stringify({ did: "did:plc:x", pds: "https://pds.allowed.test" }),
203
+
{ status: 200 },
204
+
);
205
+
}
206
+
throw new Error(`Unexpected fetch: ${url}`);
207
+
});
208
+
const db = await createTestDbWithSchema();
209
+
const client = await getClient("did:plc:x" as Did, db, {
210
+
namespace: "test",
211
+
collections: {},
212
+
networkOverrides: {
213
+
slingshotUrl: "https://my-slingshot.test/xrpc/com.bad-example.identity.resolveMiniDoc",
214
+
additionalAllowedHosts: ["pds.allowed.test"],
215
+
},
216
+
});
217
+
expect(client).toBeDefined();
218
+
const calls = fetchSpy.mock.calls.map((c) => String(c[0]));
219
+
expect(calls.some((u) => u.includes("my-slingshot.test"))).toBe(true);
220
+
});
221
+
222
+
it("getPDS with no config uses the default slingshot URL (backward-compat)", async () => {
223
+
fetchSpy.mockImplementation(async (input) => {
224
+
const url = String(input);
225
+
if (url.includes("slingshot.microcosm.blue")) {
226
+
return new Response(
227
+
JSON.stringify({ did: "did:plc:y", pds: "https://shimeji.us-east.host.bsky.network" }),
228
+
{ status: 200 },
229
+
);
230
+
}
231
+
throw new Error(`Unexpected fetch: ${url}`);
232
+
});
233
+
const db = await createTestDbWithSchema();
234
+
const pds = await getPDS("did:plc:y" as Did, db);
235
+
expect(pds).toBe("https://shimeji.us-east.host.bsky.network");
236
+
const calls = fetchSpy.mock.calls.map((c) => String(c[0]));
237
+
expect(calls.some((u) => u.includes("slingshot.microcosm.blue"))).toBe(true);
238
+
});
239
+
});
+87
packages/contrail/tests/identity-config.test.ts
+87
packages/contrail/tests/identity-config.test.ts
···
1
+
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
2
+
import {
3
+
resolveIdentity,
4
+
resolveIdentities,
5
+
resolveActor,
6
+
refreshStaleIdentities,
7
+
} from "../src/core/identity";
8
+
import { __resetPdsCachesForTests } from "../src/core/client";
9
+
import { createTestDbWithSchema } from "./helpers";
10
+
import type { Did } from "@atcute/lexicons";
11
+
12
+
describe("identity.ts — config plumb-through", () => {
13
+
let fetchSpy: ReturnType<typeof vi.spyOn>;
14
+
15
+
beforeEach(() => {
16
+
__resetPdsCachesForTests();
17
+
fetchSpy = vi.spyOn(global, "fetch");
18
+
});
19
+
20
+
afterEach(() => {
21
+
fetchSpy.mockRestore();
22
+
});
23
+
24
+
const overrideConfig = {
25
+
namespace: "test",
26
+
collections: {},
27
+
networkOverrides: {
28
+
slingshotUrl: "https://my-slingshot.test/xrpc/com.bad-example.identity.resolveMiniDoc",
29
+
additionalAllowedHosts: ["pds.allowed.test"],
30
+
},
31
+
};
32
+
33
+
it("resolveIdentity routes slingshot to override URL", async () => {
34
+
fetchSpy.mockResolvedValue(
35
+
new Response(JSON.stringify({ did: "did:plc:a", handle: "a.test", pds: "https://pds.allowed.test" }), { status: 200 }),
36
+
);
37
+
const db = await createTestDbWithSchema();
38
+
const id = await resolveIdentity(db, "did:plc:a" as Did, overrideConfig);
39
+
expect(id.pds).toBe("https://pds.allowed.test");
40
+
expect(fetchSpy.mock.calls.some(([u]) => String(u).includes("my-slingshot.test"))).toBe(true);
41
+
});
42
+
43
+
it("resolveIdentities batch routes to override URL", async () => {
44
+
fetchSpy.mockImplementation(async (input) => {
45
+
const url = String(input);
46
+
const id = new URL(url).searchParams.get("identifier") ?? "did:plc:?";
47
+
return new Response(
48
+
JSON.stringify({ did: id, handle: `${id}.test`, pds: "https://pds.allowed.test" }),
49
+
{ status: 200 },
50
+
);
51
+
});
52
+
const db = await createTestDbWithSchema();
53
+
const m = await resolveIdentities(db, ["did:plc:b", "did:plc:c"], overrideConfig);
54
+
expect(m.get("did:plc:b")?.pds).toBe("https://pds.allowed.test");
55
+
expect(m.get("did:plc:c")?.pds).toBe("https://pds.allowed.test");
56
+
});
57
+
58
+
it("resolveActor routes a handle lookup to override URL", async () => {
59
+
fetchSpy.mockResolvedValue(
60
+
new Response(JSON.stringify({ did: "did:plc:d", handle: "user.test", pds: "https://pds.allowed.test" }), { status: 200 }),
61
+
);
62
+
const db = await createTestDbWithSchema();
63
+
const did = await resolveActor(db, "user.test", overrideConfig);
64
+
expect(did).toBe("did:plc:d");
65
+
expect(fetchSpy.mock.calls.some(([u]) => String(u).includes("my-slingshot.test"))).toBe(true);
66
+
});
67
+
68
+
it("refreshStaleIdentities routes to override URL", async () => {
69
+
fetchSpy.mockResolvedValue(
70
+
new Response(JSON.stringify({ did: "did:plc:e", handle: "e.test", pds: "https://pds.allowed.test" }), { status: 200 }),
71
+
);
72
+
const db = await createTestDbWithSchema();
73
+
await refreshStaleIdentities(db, ["did:plc:e"], overrideConfig);
74
+
const row = await db.prepare("SELECT did, pds FROM identities WHERE did = ?").bind("did:plc:e").first<{ pds: string }>();
75
+
expect(row?.pds).toBe("https://pds.allowed.test");
76
+
});
77
+
78
+
it("backward-compat: no config preserves default slingshot URL", async () => {
79
+
fetchSpy.mockResolvedValue(
80
+
new Response(JSON.stringify({ did: "did:plc:f", handle: "f.bsky.social", pds: "https://shimeji.us-east.host.bsky.network" }), { status: 200 }),
81
+
);
82
+
const db = await createTestDbWithSchema();
83
+
const id = await resolveIdentity(db, "did:plc:f" as Did);
84
+
expect(id.pds).toBe("https://shimeji.us-east.host.bsky.network");
85
+
expect(fetchSpy.mock.calls.some(([u]) => String(u).includes("slingshot.microcosm.blue"))).toBe(true);
86
+
});
87
+
});
+81
packages/contrail/tests/labels-resolve.test.ts
+81
packages/contrail/tests/labels-resolve.test.ts
···
1
+
import { describe, it, expect, vi } from "vitest";
2
+
import {
3
+
CompositeDidDocumentResolver,
4
+
PlcDidDocumentResolver,
5
+
WebDidDocumentResolver,
6
+
type DidDocumentResolver,
7
+
} from "@atcute/identity-resolver";
8
+
import {
9
+
resolveLabelerEndpoint,
10
+
validateEndpointUrl,
11
+
} from "../src/core/labels/resolve";
12
+
13
+
describe("validateEndpointUrl additionalAllowedHosts", () => {
14
+
it("rejects pds.dev.svc.cluster.local without override (HTTP + private hostname)", () => {
15
+
expect(validateEndpointUrl("http://pds.dev.svc.cluster.local:2583")).toBe(false);
16
+
});
17
+
18
+
it("accepts pds.dev.svc.cluster.local when listed in additionalAllowedHosts (case-insensitive)", () => {
19
+
expect(
20
+
validateEndpointUrl("http://PDS.dev.svc.cluster.local:2583", [
21
+
"pds.dev.svc.cluster.local",
22
+
]),
23
+
).toBe(true);
24
+
});
25
+
26
+
it("does not relax HTTPS requirement for non-listed hosts when override is present", () => {
27
+
expect(
28
+
validateEndpointUrl("http://attacker.com", ["pds.dev.svc.cluster.local"]),
29
+
).toBe(false);
30
+
});
31
+
32
+
it("ignores port differences (host-only match)", () => {
33
+
expect(
34
+
validateEndpointUrl("http://pds.dev.svc.cluster.local:9999", [
35
+
"pds.dev.svc.cluster.local",
36
+
]),
37
+
).toBe(true);
38
+
});
39
+
});
40
+
41
+
describe("resolveLabelerEndpoint resolver injection", () => {
42
+
it("uses networkOverrides.resolver when provided", async () => {
43
+
const mockResolver = {
44
+
resolve: vi.fn().mockResolvedValue({
45
+
service: [
46
+
{ id: "#atproto_labeler", serviceEndpoint: "https://labeler.test" },
47
+
],
48
+
}),
49
+
};
50
+
const endpoint = await resolveLabelerEndpoint("did:plc:abc123", {
51
+
resolver: mockResolver as unknown as DidDocumentResolver,
52
+
});
53
+
expect(endpoint).toBe("https://labeler.test");
54
+
expect(mockResolver.resolve).toHaveBeenCalledOnce();
55
+
});
56
+
57
+
it("applies additionalAllowedHosts to resolved labeler endpoint", async () => {
58
+
const mockResolver = {
59
+
resolve: vi.fn().mockResolvedValue({
60
+
service: [
61
+
{
62
+
id: "#atproto_labeler",
63
+
serviceEndpoint: "http://labeler.dev.svc.cluster.local:2583",
64
+
},
65
+
],
66
+
}),
67
+
};
68
+
// Without override the http endpoint should be rejected
69
+
const rejected = await resolveLabelerEndpoint("did:plc:abc123", {
70
+
resolver: mockResolver as unknown as DidDocumentResolver,
71
+
});
72
+
expect(rejected).toBeNull();
73
+
74
+
// With override it should pass
75
+
const accepted = await resolveLabelerEndpoint("did:plc:abc123", {
76
+
resolver: mockResolver as unknown as DidDocumentResolver,
77
+
additionalAllowedHosts: ["labeler.dev.svc.cluster.local"],
78
+
});
79
+
expect(accepted).toBe("http://labeler.dev.svc.cluster.local:2583");
80
+
});
81
+
});
+96
packages/contrail/tests/network-overrides-appview.test.ts
+96
packages/contrail/tests/network-overrides-appview.test.ts
···
1
+
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
2
+
import { createApp } from "../src/core/router";
3
+
import { runIngestCycle } from "../src/core/jetstream";
4
+
import { __resetPdsCachesForTests } from "../src/core/client";
5
+
import { createTestDbWithSchema, TEST_CONFIG } from "./helpers";
6
+
import type { ContrailConfig } from "../src/core/types";
7
+
8
+
// Mock the Jetstream subscription so `runIngestCycle` ingests one synthetic
9
+
// commit without opening a real WebSocket. Everything else in the appview
10
+
// ingest path runs for real, so this exercises the live-ingest refresh path
11
+
// end-to-end (the smoking-gun call site `refreshStaleIdentities(db, dids)`).
12
+
vi.mock("@atcute/jetstream", async (importOriginal) => {
13
+
const actual = await importOriginal<typeof import("@atcute/jetstream")>();
14
+
class MockJetstreamSubscription {
15
+
cursor: number | null = null;
16
+
constructor(_opts: unknown) {}
17
+
async *[Symbol.asyncIterator]() {
18
+
yield {
19
+
kind: "commit",
20
+
// A past time_us so the ingest loop doesn't treat it as "caught up".
21
+
time_us: 1_000_000,
22
+
did: "did:plc:ingest",
23
+
commit: {
24
+
collection: "community.lexicon.calendar.event",
25
+
operation: "create",
26
+
rkey: "abc",
27
+
cid: "bafyabc",
28
+
record: { name: "Test Event", startsAt: "2026-04-01T10:00:00Z", mode: "online" },
29
+
},
30
+
};
31
+
}
32
+
}
33
+
return { ...actual, JetstreamSubscription: MockJetstreamSubscription };
34
+
});
35
+
36
+
const silentLogger = { log() {}, warn() {}, error() {} };
37
+
38
+
const OVERRIDE = {
39
+
slingshotUrl: "https://my-slingshot.test/xrpc/com.bad-example.identity.resolveMiniDoc",
40
+
additionalAllowedHosts: ["pds.allowed.test"],
41
+
};
42
+
43
+
function overrideConfig(): ContrailConfig {
44
+
return { ...TEST_CONFIG, logger: silentLogger, networkOverrides: OVERRIDE };
45
+
}
46
+
47
+
describe("networkOverrides — appview entry points thread config", () => {
48
+
let fetchSpy: ReturnType<typeof vi.spyOn>;
49
+
50
+
beforeEach(() => {
51
+
__resetPdsCachesForTests();
52
+
fetchSpy = vi.spyOn(globalThis, "fetch").mockResolvedValue(
53
+
new Response(
54
+
JSON.stringify({ did: "did:plc:resolved", handle: "user.test", pds: "https://pds.allowed.test" }),
55
+
{ status: 200, headers: { "content-type": "application/json" } },
56
+
),
57
+
);
58
+
});
59
+
60
+
afterEach(() => {
61
+
fetchSpy.mockRestore();
62
+
});
63
+
64
+
// router actor-resolution path: GET /xrpc/<ns>.getProfile -> resolveActor
65
+
it("router getProfile resolves a handle via the override slingshot", async () => {
66
+
const db = await createTestDbWithSchema();
67
+
const app = createApp(db, overrideConfig());
68
+
69
+
await app.fetch(
70
+
new Request(`http://localhost/xrpc/${TEST_CONFIG.namespace}.getProfile?actor=user.test`),
71
+
);
72
+
73
+
// The handle lookup (identifier=user.test) must go to the override
74
+
// slingshot, not the default public one. Before the fix `resolveActor`
75
+
// was called without `config`, so this fetch hit the default URL.
76
+
const hitOverrideForHandle = fetchSpy.mock.calls.some(([u]) => {
77
+
const s = String(u);
78
+
return s.includes("my-slingshot.test") && s.includes("identifier=user.test");
79
+
});
80
+
expect(hitOverrideForHandle).toBe(true);
81
+
});
82
+
83
+
// live-ingest refresh path: runIngestCycle -> refreshStaleIdentities
84
+
it("jetstream ingest cycle refreshes identities via the override slingshot", async () => {
85
+
const db = await createTestDbWithSchema();
86
+
87
+
await runIngestCycle(db, overrideConfig(), 1_000);
88
+
89
+
// refreshStaleIdentities resolved the ingested DID; before the fix it was
90
+
// called without `config`, so the resolve hit the default slingshot.
91
+
const hitOverride = fetchSpy.mock.calls.some(([u]) =>
92
+
String(u).includes("my-slingshot.test"),
93
+
);
94
+
expect(hitOverride).toBe(true);
95
+
});
96
+
});
+189
packages/contrail/tests/network-overrides.test.ts
+189
packages/contrail/tests/network-overrides.test.ts
···
1
+
import { describe, it, expect, beforeAll, afterAll, beforeEach } from "vitest";
2
+
import http from "node:http";
3
+
import type { AddressInfo } from "node:net";
4
+
import { resolvePDS, getClient, __resetPdsCachesForTests } from "../src/core/client";
5
+
import { refreshStaleIdentities } from "../src/core/identity";
6
+
import { createTestDbWithSchema } from "./helpers";
7
+
import type { ContrailConfig } from "../src/core/types";
8
+
import type { Did } from "@atcute/lexicons";
9
+
import {
10
+
CompositeDidDocumentResolver,
11
+
PlcDidDocumentResolver,
12
+
WebDidDocumentResolver,
13
+
} from "@atcute/identity-resolver";
14
+
15
+
type Hit = { method: string; url: string };
16
+
17
+
interface Stub {
18
+
url: string;
19
+
hits: Hit[];
20
+
setHandler: (h: (req: http.IncomingMessage, res: http.ServerResponse) => void) => void;
21
+
close: () => Promise<void>;
22
+
}
23
+
24
+
async function startStub(): Promise<Stub> {
25
+
const hits: Hit[] = [];
26
+
let handler: (req: http.IncomingMessage, res: http.ServerResponse) => void = (_req, res) => {
27
+
res.writeHead(404);
28
+
res.end();
29
+
};
30
+
const server = http.createServer((req, res) => {
31
+
hits.push({ method: req.method ?? "?", url: req.url ?? "" });
32
+
handler(req, res);
33
+
});
34
+
await new Promise<void>((r) => server.listen(0, "127.0.0.1", r));
35
+
const { port } = server.address() as AddressInfo;
36
+
return {
37
+
url: `http://127.0.0.1:${port}`,
38
+
hits,
39
+
setHandler: (h) => {
40
+
handler = h;
41
+
},
42
+
close: () => new Promise<void>((r) => server.close(() => r())),
43
+
};
44
+
}
45
+
46
+
let plc: Stub;
47
+
let slingshot: Stub;
48
+
49
+
const baseConfig: ContrailConfig = {
50
+
namespace: "test",
51
+
collections: {},
52
+
};
53
+
54
+
beforeAll(async () => {
55
+
plc = await startStub();
56
+
slingshot = await startStub();
57
+
});
58
+
59
+
afterAll(async () => {
60
+
await plc.close();
61
+
await slingshot.close();
62
+
});
63
+
64
+
beforeEach(() => {
65
+
plc.hits.length = 0;
66
+
slingshot.hits.length = 0;
67
+
__resetPdsCachesForTests();
68
+
});
69
+
70
+
describe("networkOverrides — full chain integration", () => {
71
+
it("resolvePDS routes slingshot fetch to slingshotUrl override", async () => {
72
+
slingshot.setHandler((_req, res) => {
73
+
res.writeHead(200, { "content-type": "application/json" });
74
+
res.end(JSON.stringify({ did: "did:plc:abc", handle: "alice.test", pds: "http://pds.private.test" }));
75
+
});
76
+
77
+
const config: ContrailConfig = {
78
+
...baseConfig,
79
+
networkOverrides: {
80
+
slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`,
81
+
additionalAllowedHosts: ["pds.private.test"],
82
+
},
83
+
};
84
+
const result = await resolvePDS("did:plc:abc", config);
85
+
expect(result?.pds).toBe("http://pds.private.test");
86
+
expect(slingshot.hits.length).toBeGreaterThan(0);
87
+
expect(slingshot.hits[0].url).toContain("identifier=did%3Aplc%3Aabc");
88
+
});
89
+
90
+
it("rejects pds when not in allowlist (default validator still applies)", async () => {
91
+
slingshot.setHandler((_req, res) => {
92
+
res.writeHead(200, { "content-type": "application/json" });
93
+
res.end(JSON.stringify({ did: "did:plc:bcd", pds: "http://pds.private.test" }));
94
+
});
95
+
96
+
const config: ContrailConfig = {
97
+
...baseConfig,
98
+
networkOverrides: {
99
+
slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`,
100
+
},
101
+
};
102
+
const result = await resolvePDS("did:plc:bcd", config);
103
+
expect(result?.pds).toBe(null);
104
+
});
105
+
106
+
it("falls back to plcUrl when slingshot returns no pds", async () => {
107
+
slingshot.setHandler((_req, res) => {
108
+
res.writeHead(200, { "content-type": "application/json" });
109
+
res.end(JSON.stringify({ did: "did:plc:cde", handle: "carol.test" }));
110
+
});
111
+
plc.setHandler((req, res) => {
112
+
const decoded = req.url ? decodeURIComponent(req.url) : "";
113
+
if (decoded.includes("did:plc:cde")) {
114
+
res.writeHead(200, { "content-type": "application/did+ld+json" });
115
+
res.end(JSON.stringify({
116
+
"@context": ["https://www.w3.org/ns/did/v1"],
117
+
id: "did:plc:cde",
118
+
alsoKnownAs: ["at://carol.test"],
119
+
verificationMethod: [],
120
+
service: [
121
+
{ id: "#atproto_pds", type: "AtprotoPersonalDataServer", serviceEndpoint: "http://pds.private.test" },
122
+
],
123
+
}));
124
+
} else {
125
+
res.writeHead(404);
126
+
res.end();
127
+
}
128
+
});
129
+
130
+
const resolver = new CompositeDidDocumentResolver({
131
+
methods: {
132
+
plc: new PlcDidDocumentResolver({ apiUrl: plc.url }),
133
+
web: new WebDidDocumentResolver(),
134
+
},
135
+
});
136
+
const config: ContrailConfig = {
137
+
...baseConfig,
138
+
networkOverrides: {
139
+
slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`,
140
+
resolver,
141
+
additionalAllowedHosts: ["pds.private.test"],
142
+
},
143
+
};
144
+
const result = await resolvePDS("did:plc:cde", config);
145
+
expect(result?.pds).toBe("http://pds.private.test");
146
+
expect(plc.hits.length).toBeGreaterThan(0);
147
+
});
148
+
149
+
it("refreshStaleIdentities persists pds via override path", async () => {
150
+
slingshot.setHandler((_req, res) => {
151
+
res.writeHead(200, { "content-type": "application/json" });
152
+
res.end(JSON.stringify({ did: "did:plc:def", handle: "dave.test", pds: "http://pds.private.test" }));
153
+
});
154
+
const config: ContrailConfig = {
155
+
...baseConfig,
156
+
networkOverrides: {
157
+
slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`,
158
+
additionalAllowedHosts: ["pds.private.test"],
159
+
},
160
+
};
161
+
const db = await createTestDbWithSchema();
162
+
await refreshStaleIdentities(db, ["did:plc:def"], config);
163
+
164
+
const row = await db
165
+
.prepare("SELECT did, pds FROM identities WHERE did = ?")
166
+
.bind("did:plc:def")
167
+
.first<{ did: string; pds: string }>();
168
+
expect(row?.pds).toBe("http://pds.private.test");
169
+
expect(slingshot.hits.length).toBeGreaterThan(0);
170
+
});
171
+
172
+
it("getClient with override config resolves PDS via stubbed slingshot", async () => {
173
+
slingshot.setHandler((_req, res) => {
174
+
res.writeHead(200, { "content-type": "application/json" });
175
+
res.end(JSON.stringify({ did: "did:plc:efg", handle: "eve.test", pds: "http://pds.private.test" }));
176
+
});
177
+
const config: ContrailConfig = {
178
+
...baseConfig,
179
+
networkOverrides: {
180
+
slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`,
181
+
additionalAllowedHosts: ["pds.private.test"],
182
+
},
183
+
};
184
+
const db = await createTestDbWithSchema();
185
+
const client = await getClient("did:plc:efg" as Did, db, config);
186
+
expect(client).toBeDefined();
187
+
expect(slingshot.hits.some((h) => h.url.includes("identifier=did%3Aplc%3Aefg"))).toBe(true);
188
+
});
189
+
});
+92
packages/contrail/tests/postgres-concurrent-init.test.ts
+92
packages/contrail/tests/postgres-concurrent-init.test.ts
···
1
+
import { describe, it, expect, beforeAll, afterAll, beforeEach } from "vitest";
2
+
import pg from "pg";
3
+
import { createPostgresDatabase } from "../src/adapters/postgres";
4
+
import { initSchema } from "../src/core/db/schema";
5
+
import { resolveConfig } from "../src/core/types";
6
+
7
+
/**
8
+
* Postgres-dialect concurrent-init race.
9
+
*
10
+
* SQLite serializes DDL globally, so the existing `schema-idempotency.test.ts`
11
+
* (which uses `createSqliteDatabase`) can't surface the Postgres-specific race
12
+
* where two concurrent `CREATE TABLE IF NOT EXISTS` statements both pass the
13
+
* existence check and then both try to insert into pg_class/pg_type, with the
14
+
* loser raising 23505 on `pg_type_typname_nsp_index` (the unique index on
15
+
* (typname, typnamespace)).
16
+
*
17
+
* Real-world hit: discovered during PR44 Phase C local validation when the OM
18
+
* API consumer's `contrail-init-idempotency.spec.ts` ran three parallel
19
+
* `contrail.init(db)` calls against a fresh Postgres schema.
20
+
*/
21
+
22
+
const TEST_CONFIG = resolveConfig({
23
+
namespace: "com.example",
24
+
collections: {
25
+
event: {
26
+
collection: "community.lexicon.calendar.event",
27
+
queryable: { mode: {}, name: {}, startsAt: { type: "range" } },
28
+
searchable: ["name", "description"],
29
+
relations: {
30
+
rsvps: {
31
+
collection: "rsvp",
32
+
groupBy: "status",
33
+
groups: {
34
+
going: "community.lexicon.calendar.rsvp#going",
35
+
},
36
+
},
37
+
},
38
+
},
39
+
rsvp: {
40
+
collection: "community.lexicon.calendar.rsvp",
41
+
references: {
42
+
event: {
43
+
collection: "event",
44
+
field: "subject.uri",
45
+
},
46
+
},
47
+
},
48
+
},
49
+
});
50
+
51
+
const PG_URL = process.env.TEST_DATABASE_URL;
52
+
if (!PG_URL) {
53
+
describe.skip("PostgreSQL concurrent init (TEST_DATABASE_URL not set)", () => {
54
+
it("skipped", () => {});
55
+
});
56
+
} else {
57
+
let pool: pg.Pool;
58
+
let db: ReturnType<typeof createPostgresDatabase>;
59
+
60
+
beforeAll(async () => {
61
+
pool = new pg.Pool({ connectionString: PG_URL });
62
+
await pool.query("SELECT 1");
63
+
db = createPostgresDatabase(pool);
64
+
});
65
+
66
+
afterAll(async () => {
67
+
await pool?.end();
68
+
});
69
+
70
+
beforeEach(async () => {
71
+
const tables = await pool.query(
72
+
`SELECT tablename FROM pg_tables WHERE schemaname = 'public'
73
+
AND (tablename LIKE 'records_%' OR tablename LIKE 'fts_%'
74
+
OR tablename IN ('backfills', 'discovery', 'cursor', 'identities', 'feed_items', 'feed_backfills'))`
75
+
);
76
+
for (const { tablename } of tables.rows) {
77
+
await pool.query(`DROP TABLE IF EXISTS ${tablename} CASCADE`);
78
+
}
79
+
});
80
+
81
+
describe("PostgreSQL initSchema under concurrency", () => {
82
+
it("is safe to call three times concurrently against a fresh schema", async () => {
83
+
await expect(
84
+
Promise.all([
85
+
initSchema(db, TEST_CONFIG),
86
+
initSchema(db, TEST_CONFIG),
87
+
initSchema(db, TEST_CONFIG),
88
+
])
89
+
).resolves.not.toThrow();
90
+
});
91
+
});
92
+
}
+139
packages/contrail/tests/schema-idempotency.test.ts
+139
packages/contrail/tests/schema-idempotency.test.ts
···
1
+
import { describe, it, expect } from "vitest";
2
+
import { initSchema, addColumnIfNotExists } from "../src/core/db/schema";
3
+
import type { Database } from "../src/core/types";
4
+
import { resolveConfig } from "../src/core/types";
5
+
import { createTestDb, TEST_CONFIG } from "./helpers";
6
+
7
+
/**
8
+
* L3 — schema idempotency.
9
+
*
10
+
* These tests pin the contract that `initSchema` is safe to call repeatedly
11
+
* (sequentially or concurrently) and that real DDL errors are NOT silently
12
+
* swallowed. The previous implementation wrapped ALTER TABLE statements in
13
+
* `try { ... } catch { /* ignore *\/ }` blocks that masked syntax errors,
14
+
* missing tables, and type mismatches alongside the intended duplicate-column
15
+
* case. After L3, only duplicate-column races are absorbed (via dialect-aware
16
+
* IF-NOT-EXISTS / PRAGMA pre-check); other DDL failures surface.
17
+
*/
18
+
19
+
describe("initSchema idempotency", () => {
20
+
it("is safe to call twice sequentially", async () => {
21
+
const db = createTestDb();
22
+
await initSchema(db, TEST_CONFIG);
23
+
await expect(initSchema(db, TEST_CONFIG)).resolves.not.toThrow();
24
+
});
25
+
26
+
it("is safe to call concurrently", async () => {
27
+
const db = createTestDb();
28
+
await Promise.all([
29
+
initSchema(db, TEST_CONFIG),
30
+
initSchema(db, TEST_CONFIG),
31
+
initSchema(db, TEST_CONFIG),
32
+
]);
33
+
34
+
// Verify count column on records_event exists exactly once. The count
35
+
// columns are added via ALTER TABLE; duplicate adds would have failed
36
+
// without idempotent ALTER.
37
+
const cols = await db
38
+
.prepare("PRAGMA table_info(records_event)")
39
+
.all<{ name: string }>();
40
+
const names = cols.results.map((c) => c.name);
41
+
42
+
// There should be exactly one of each grouped-count column (sanity check
43
+
// that no parallel run got further than the first).
44
+
const countCols = names.filter((n) => n.startsWith("count_"));
45
+
const dedup = new Set(countCols);
46
+
expect(countCols.length).toBe(dedup.size);
47
+
// And we should have at least one count column (config defines rsvp groups).
48
+
expect(countCols.length).toBeGreaterThan(0);
49
+
});
50
+
51
+
it("does not leave duplicate count columns after repeated init", async () => {
52
+
const db = createTestDb();
53
+
for (let i = 0; i < 5; i++) {
54
+
await initSchema(db, TEST_CONFIG);
55
+
}
56
+
const cols = await db
57
+
.prepare("PRAGMA table_info(records_event)")
58
+
.all<{ name: string }>();
59
+
const names = cols.results.map((c) => c.name);
60
+
expect(new Set(names).size).toBe(names.length);
61
+
});
62
+
});
63
+
64
+
describe("addColumnIfNotExists", () => {
65
+
// The helper is the seam where dialect-aware idempotent ALTER lives. We
66
+
// verify it both adds a column when absent and is a no-op when present.
67
+
it("adds the column when absent", async () => {
68
+
const db = createTestDb();
69
+
await db.prepare("CREATE TABLE t (id INTEGER PRIMARY KEY)").run();
70
+
71
+
await addColumnIfNotExists(db, "t", "extra", "INTEGER NOT NULL DEFAULT 0");
72
+
73
+
const cols = await db.prepare("PRAGMA table_info(t)").all<{ name: string }>();
74
+
expect(cols.results.map((c) => c.name)).toContain("extra");
75
+
});
76
+
77
+
it("is a no-op when the column already exists", async () => {
78
+
const db = createTestDb();
79
+
await db.prepare("CREATE TABLE t (id INTEGER PRIMARY KEY, extra INTEGER)").run();
80
+
81
+
// First call: column already present — should not throw.
82
+
await expect(
83
+
addColumnIfNotExists(db, "t", "extra", "INTEGER NOT NULL DEFAULT 0")
84
+
).resolves.not.toThrow();
85
+
86
+
// Calling it again should still be a no-op.
87
+
await expect(
88
+
addColumnIfNotExists(db, "t", "extra", "INTEGER NOT NULL DEFAULT 0")
89
+
).resolves.not.toThrow();
90
+
91
+
// And we still have exactly one `extra` column.
92
+
const cols = await db.prepare("PRAGMA table_info(t)").all<{ name: string }>();
93
+
const extras = cols.results.filter((c) => c.name === "extra");
94
+
expect(extras.length).toBe(1);
95
+
});
96
+
97
+
it("surfaces real DDL errors (target table does not exist)", async () => {
98
+
// The previous swallow-all `try { ... } catch { /* ignore */ }` masked
99
+
// *any* DDL failure, including target-table-missing. The new helper must
100
+
// only absorb the duplicate-column case and surface everything else.
101
+
const db = createTestDb();
102
+
103
+
await expect(
104
+
addColumnIfNotExists(db, "no_such_table", "x", "INTEGER")
105
+
).rejects.toThrow();
106
+
});
107
+
});
108
+
109
+
describe("initSchema with extra schemas — real DDL errors surface", () => {
110
+
// Belt-and-suspenders: confirm that a genuine failure inside an extension
111
+
// schema module propagates rather than being absorbed. This is the
112
+
// user-visible behavior change from L3.
113
+
it("propagates errors thrown from an extra schema", async () => {
114
+
const db = createTestDb();
115
+
const broken = async (_db: Database) => {
116
+
throw new Error("synthetic DDL failure");
117
+
};
118
+
119
+
await expect(
120
+
initSchema(db, TEST_CONFIG, { extraSchemas: [broken] })
121
+
).rejects.toThrow("synthetic DDL failure");
122
+
});
123
+
124
+
it("treats a config without spaces/labels/feeds the same way (sanity)", async () => {
125
+
// Minimal config with no relations — should still init cleanly twice.
126
+
const minimal = resolveConfig({
127
+
namespace: "com.example",
128
+
collections: {
129
+
foo: {
130
+
collection: "com.example.foo",
131
+
queryable: {},
132
+
},
133
+
},
134
+
});
135
+
const db = createTestDb();
136
+
await initSchema(db, minimal);
137
+
await expect(initSchema(db, minimal)).resolves.not.toThrow();
138
+
});
139
+
});
+65
packages/contrail/tests/spaces-auth.test.ts
+65
packages/contrail/tests/spaces-auth.test.ts
···
1
+
import { describe, it, expect } from "vitest";
2
+
import {
3
+
PlcDidDocumentResolver,
4
+
CompositeDidDocumentResolver,
5
+
WebDidDocumentResolver,
6
+
} from "@atcute/identity-resolver";
7
+
import { buildVerifier } from "../src/core/spaces/auth";
8
+
import type { AuthorityConfig } from "../src/core/spaces/types";
9
+
10
+
const CUSTOM_PLC = "http://custom-plc.test";
11
+
12
+
function makeAuthority(overrides: Partial<AuthorityConfig> = {}): AuthorityConfig {
13
+
return {
14
+
type: "tools.atmo.event.space",
15
+
serviceDid: "did:web:authority.test",
16
+
...overrides,
17
+
} as AuthorityConfig;
18
+
}
19
+
20
+
// ServiceJwtVerifier from @atcute/xrpc-server@0.1.12 exposes the resolver as
21
+
// the public instance field `didDocResolver` (verified against
22
+
// node_modules/.../auth/jwt-verifier.d.ts). The plan's hint at `.resolver` was
23
+
// a guess — we use the real field here so the test verifies the resolver
24
+
// actually wired into the verifier instance.
25
+
describe("buildVerifier resolver precedence", () => {
26
+
it("uses AuthorityConfig.resolver when provided (most-specific wins)", () => {
27
+
const specific = new CompositeDidDocumentResolver({
28
+
methods: {
29
+
plc: new PlcDidDocumentResolver({ apiUrl: "http://specific.test" }),
30
+
web: new WebDidDocumentResolver(),
31
+
},
32
+
});
33
+
const network = new CompositeDidDocumentResolver({
34
+
methods: {
35
+
plc: new PlcDidDocumentResolver({ apiUrl: CUSTOM_PLC }),
36
+
web: new WebDidDocumentResolver(),
37
+
},
38
+
});
39
+
const verifier = buildVerifier(makeAuthority({ resolver: specific }), {
40
+
resolver: network,
41
+
});
42
+
expect(verifier.didDocResolver).toBe(specific);
43
+
});
44
+
45
+
it("falls back to networkOverrides.resolver when authority resolver is absent", () => {
46
+
const network = new CompositeDidDocumentResolver({
47
+
methods: {
48
+
plc: new PlcDidDocumentResolver({ apiUrl: CUSTOM_PLC }),
49
+
web: new WebDidDocumentResolver(),
50
+
},
51
+
});
52
+
const verifier = buildVerifier(makeAuthority(), { resolver: network });
53
+
expect(verifier.didDocResolver).toBe(network);
54
+
});
55
+
56
+
it("falls back to default composite when both are absent", () => {
57
+
const verifier = buildVerifier(makeAuthority(), {});
58
+
expect(verifier.didDocResolver).toBeInstanceOf(CompositeDidDocumentResolver);
59
+
});
60
+
61
+
it("treats omitted second arg the same as empty networkOverrides (backward-compat)", () => {
62
+
const verifier = buildVerifier(makeAuthority());
63
+
expect(verifier.didDocResolver).toBeInstanceOf(CompositeDidDocumentResolver);
64
+
});
65
+
});
+4
-4
packages/contrail-appview/src/core/backfill.ts
+4
-4
packages/contrail-appview/src/core/backfill.ts
···
186
186
if (!client) {
187
187
try {
188
188
client = await withRetry(
189
-
() => getClient(did as Did, db),
189
+
() => getClient(did as Did, db, config),
190
190
`getClient(${did})`,
191
191
Math.min(retries, 1),
192
192
timeout
···
356
356
for (let i = 0; i < dids.length; i += 200) {
357
357
await Promise.allSettled(
358
358
dids.slice(i, i + 200).map((did) =>
359
-
getPDS(did as Did, db).catch(() => {})
359
+
getPDS(did as Did, db, config).catch(() => {})
360
360
)
361
361
);
362
362
}
···
386
386
let client: Client | undefined;
387
387
try {
388
388
client = await withRetry(
389
-
() => getClient(did as Did, db),
389
+
() => getClient(did as Did, db, config),
390
390
`getClient(${did})`,
391
391
0,
392
392
FAST_TIMEOUT
···
436
436
let client: Client | undefined;
437
437
try {
438
438
client = await withRetry(
439
-
() => getClient(did as Did, db),
439
+
() => getClient(did as Did, db, config),
440
440
`getClient(${did})`,
441
441
2
442
442
);
+1
-1
packages/contrail-appview/src/core/jetstream.ts
+1
-1
packages/contrail-appview/src/core/jetstream.ts
···
307
307
const uniqueDids = [...new Set(events.map((e) => e.did))];
308
308
if (uniqueDids.length > 0) {
309
309
try {
310
-
await refreshStaleIdentities(db, uniqueDids);
310
+
await refreshStaleIdentities(db, uniqueDids, config);
311
311
} catch (err) {
312
312
log.warn(`Identity refresh failed: ${err}`);
313
313
}
+1
-1
packages/contrail-appview/src/core/persistent.ts
+1
-1
packages/contrail-appview/src/core/persistent.ts
···
157
157
const uniqueDids = [...new Set(batch.map((e) => e.did))];
158
158
if (uniqueDids.length > 0) {
159
159
try {
160
-
await refreshStaleIdentities(db, uniqueDids);
160
+
await refreshStaleIdentities(db, uniqueDids, config);
161
161
} catch (err) {
162
162
log.warn(`Identity refresh failed: ${err}`);
163
163
}
+1
-1
packages/contrail-appview/src/core/refresh.ts
+1
-1
packages/contrail-appview/src/core/refresh.ts
+14
-4
packages/contrail-base/src/spaces/auth.ts
+14
-4
packages/contrail-base/src/spaces/auth.ts
···
12
12
13
13
export { ServiceJwtVerifier };
14
14
15
-
/** Build a ServiceJwtVerifier from an AuthorityConfig, using the configured
16
-
* resolver or a default PLC+Web composite. The verifier checks that incoming
17
-
* JWTs target this authority's serviceDid (aud claim). */
18
-
export function buildVerifier(authority: AuthorityConfig): ServiceJwtVerifier {
15
+
/** Build a ServiceJwtVerifier from an AuthorityConfig, using (in precedence
16
+
* order) the authority-specific resolver, then the deployment-wide
17
+
* `networkOverrides.resolver`, then a default PLC+Web composite. The verifier
18
+
* checks that incoming JWTs target this authority's serviceDid (aud claim).
19
+
*
20
+
* `networkOverrides` is optional and is the same shape carried on
21
+
* `ContrailConfig.networkOverrides` — callers with a ContrailConfig in scope
22
+
* should pass `config.networkOverrides` so private-network deployments share
23
+
* one resolver across both identity resolution and service-auth verification. */
24
+
export function buildVerifier(
25
+
authority: AuthorityConfig,
26
+
networkOverrides?: { resolver?: DidDocumentResolver },
27
+
): ServiceJwtVerifier {
19
28
const resolver =
20
29
authority.resolver ??
30
+
networkOverrides?.resolver ??
21
31
new CompositeDidDocumentResolver({
22
32
methods: {
23
33
plc: new PlcDidDocumentResolver(),
+205
-29
packages/contrail-appview/src/core/db/schema.ts
+205
-29
packages/contrail-appview/src/core/db/schema.ts
···
1
1
import type { ContrailConfig, Database, ResolvedContrailConfig, ResolvedMaps } from "../types";
2
2
import type { SqlDialect } from "../dialect";
3
-
import { buildFtsSchema, getDialect } from "../dialect";
3
+
import { buildFtsSchema, getDialect, postgresDialect } from "../dialect";
4
4
import {
5
5
getRelationField,
6
6
countColumnName,
···
197
197
return stmts;
198
198
}
199
199
200
+
/**
201
+
* Idempotently add a column to a table, surfacing real DDL errors.
202
+
*
203
+
* Postgres supports `ALTER TABLE ... ADD COLUMN IF NOT EXISTS` natively, so we
204
+
* issue that and let any non-duplicate error propagate. SQLite (including
205
+
* `node:sqlite`) does NOT support `IF NOT EXISTS` on `ADD COLUMN`, so we
206
+
* pre-check `PRAGMA table_info` and short-circuit if the column is already
207
+
* there. Because the PRAGMA-check + ALTER pair is not atomic, a concurrent
208
+
* second `initSchema` call can still hit a "duplicate column name" race; we
209
+
* narrowly absorb exactly that error message and re-throw everything else.
210
+
*
211
+
* Net effect: only the duplicate-column case is absorbed. Missing tables,
212
+
* syntax errors, type mismatches, and any other DDL failure will throw.
213
+
*
214
+
* Exported for direct testing of the idempotency contract; callers in
215
+
* `initSchema` use this internally.
216
+
*/
217
+
export async function addColumnIfNotExists(
218
+
db: Database,
219
+
table: string,
220
+
column: string,
221
+
columnDef: string,
222
+
): Promise<void> {
223
+
const dialect = getDialect(db);
224
+
if (dialect === postgresDialect) {
225
+
await db.prepare(
226
+
`ALTER TABLE ${table} ADD COLUMN IF NOT EXISTS ${column} ${columnDef}`,
227
+
).run();
228
+
return;
229
+
}
230
+
// SQLite path: check existence first, then ALTER without IF NOT EXISTS.
231
+
// PRAGMA table_info() does not accept parameter binding, so we rely on the
232
+
// caller to pass a sanitized identifier (all current callers do — table
233
+
// names come from `recordsTableName`/`spacesRecordsTableName` which
234
+
// sanitize, and column names come from `countColumnName` /
235
+
// `groupedCountColumnName` which also sanitize).
236
+
const info = await db
237
+
.prepare(`PRAGMA table_info(${table})`)
238
+
.all<{ name: string }>();
239
+
if (info.results.some((c) => c.name === column)) return;
240
+
try {
241
+
await db.prepare(
242
+
`ALTER TABLE ${table} ADD COLUMN ${column} ${columnDef}`,
243
+
).run();
244
+
} catch (err) {
245
+
// Narrow swallow: only the "duplicate column" race between the PRAGMA
246
+
// read and the ALTER is acceptable. Everything else surfaces.
247
+
if (!isDuplicateColumnError(err)) throw err;
248
+
}
249
+
}
250
+
251
+
function isDuplicateColumnError(err: unknown): boolean {
252
+
if (!err || typeof err !== "object") return false;
253
+
const msg = (err as { message?: unknown }).message;
254
+
if (typeof msg !== "string") return false;
255
+
// node:sqlite / better-sqlite3: "duplicate column name: <col>"
256
+
return /duplicate column name/i.test(msg);
257
+
}
258
+
259
+
/**
260
+
* Postgres `CREATE TABLE IF NOT EXISTS` and `CREATE INDEX IF NOT EXISTS` are
261
+
* NOT atomic against concurrent creators: two transactions can both pass the
262
+
* existence check before either has inserted into `pg_class` / `pg_type`. The
263
+
* loser raises 23505 on `pg_type_typname_nsp_index` (the unique index on
264
+
* `(typname, typnamespace)`) or `pg_class_relname_nsp_index`. Pre-existing
265
+
* tables also surface as 42P07 (`duplicate_table`).
266
+
*
267
+
* SQLite serializes DDL globally, so this race never manifests there.
268
+
*
269
+
* The caller is expected to issue idempotent DDL (IF NOT EXISTS); this helper
270
+
* only absorbs the narrow concurrent-create race.
271
+
*/
272
+
function isConcurrentCreateError(err: unknown): boolean {
273
+
if (!err || typeof err !== "object") return false;
274
+
const code = (err as { code?: unknown }).code;
275
+
if (code === "42P07" || code === "42P06") return true;
276
+
if (code === "23505") {
277
+
const constraint = (err as { constraint?: unknown }).constraint;
278
+
return (
279
+
constraint === "pg_type_typname_nsp_index" ||
280
+
constraint === "pg_class_relname_nsp_index" ||
281
+
constraint === "pg_namespace_nspname_index"
282
+
);
283
+
}
284
+
return false;
285
+
}
286
+
287
+
/**
288
+
* Run a single DDL statement, absorbing only the concurrent-create race that
289
+
* `CREATE TABLE IF NOT EXISTS` / `CREATE INDEX IF NOT EXISTS` can hit on
290
+
* Postgres when multiple processes init the same schema in parallel. Genuine
291
+
* DDL errors (syntax, type mismatch, missing column) surface unchanged.
292
+
*/
293
+
async function runIdempotentDdl(db: Database, stmt: string): Promise<void> {
294
+
try {
295
+
await db.prepare(stmt).run();
296
+
} catch (err) {
297
+
if (!isConcurrentCreateError(err)) throw err;
298
+
}
299
+
}
300
+
301
+
/**
302
+
* Apply the ALTER+INDEX statements emitted by `buildCountColumns`
303
+
* idempotently and without swallowing non-duplicate errors.
304
+
*
305
+
* `buildCountColumns` mixes two statement shapes: `ALTER TABLE ... ADD COLUMN
306
+
* ...` (not idempotent on SQLite without a pre-check; supports IF NOT EXISTS
307
+
* on Postgres) and `CREATE INDEX IF NOT EXISTS ...` (idempotent on both
308
+
* dialects). We route ALTERs through `addColumnIfNotExists` and run indexes
309
+
* directly.
310
+
*/
311
+
export async function applyCountColumns(
312
+
db: Database,
313
+
config: ContrailConfig,
314
+
opts: BuilderOpts = {},
315
+
): Promise<void> {
316
+
for (const stmt of buildCountColumns(config, opts)) {
317
+
const match = stmt.match(
318
+
/^ALTER TABLE\s+(\S+)\s+ADD COLUMN\s+(\S+)\s+(.+)$/i,
319
+
);
320
+
if (match) {
321
+
const [, table, column, columnDef] = match;
322
+
await addColumnIfNotExists(db, table, column, columnDef);
323
+
} else {
324
+
await db.prepare(stmt).run();
325
+
}
326
+
}
327
+
}
328
+
200
329
function buildFeedTables(config: ContrailConfig, dialect: SqlDialect): string[] {
201
330
if (!config.feeds || Object.keys(config.feeds).length === 0) return [];
202
331
const stmts = [
···
250
379
return stmts;
251
380
}
252
381
253
-
const MIGRATIONS = [
254
-
"ALTER TABLE backfills ADD COLUMN retries INTEGER NOT NULL DEFAULT 0",
255
-
"ALTER TABLE backfills ADD COLUMN last_error TEXT",
256
-
"ALTER TABLE spaces_invites ADD COLUMN kind TEXT NOT NULL DEFAULT 'join'",
257
-
"ALTER TABLE feed_backfills ADD COLUMN retries INTEGER NOT NULL DEFAULT 0",
258
-
"ALTER TABLE feed_backfills ADD COLUMN last_error TEXT",
259
-
"ALTER TABLE feed_backfills ADD COLUMN started_at BIGINT",
382
+
/**
383
+
* Schema migrations expressed as structured ADD-COLUMN ops. Each entry is
384
+
* applied via `addColumnIfNotExists` so the operation is idempotent on both
385
+
* dialects without swallowing genuine DDL errors.
386
+
*
387
+
* `target: "spaces"` is routed to the spaces DB (which may differ from the
388
+
* main DB in split-DB deployments) and only applied when spaces is enabled.
389
+
* `target: "feeds"` is only applied when feeds are configured (the
390
+
* `feed_backfills` table doesn't exist otherwise). All other migrations
391
+
* target the main DB unconditionally.
392
+
*/
393
+
interface MigrationOp {
394
+
table: string;
395
+
column: string;
396
+
columnDef: string;
397
+
target?: "spaces" | "feeds";
398
+
}
399
+
400
+
const MIGRATIONS: MigrationOp[] = [
401
+
{ table: "backfills", column: "retries", columnDef: "INTEGER NOT NULL DEFAULT 0" },
402
+
{ table: "backfills", column: "last_error", columnDef: "TEXT" },
403
+
{
404
+
table: "spaces_invites",
405
+
column: "kind",
406
+
columnDef: "TEXT NOT NULL DEFAULT 'join'",
407
+
target: "spaces",
408
+
},
409
+
{ table: "feed_backfills", column: "retries", columnDef: "INTEGER NOT NULL DEFAULT 0", target: "feeds" },
410
+
{ table: "feed_backfills", column: "last_error", columnDef: "TEXT", target: "feeds" },
411
+
{ table: "feed_backfills", column: "started_at", columnDef: "BIGINT", target: "feeds" },
260
412
];
261
413
262
-
async function runMigrations(db: Database): Promise<void> {
263
-
for (const sql of MIGRATIONS) {
264
-
try {
265
-
await db.prepare(sql).run();
266
-
} catch {
267
-
// Column already exists — ignore
414
+
async function runMigrations(
415
+
db: Database,
416
+
spacesDb: Database | undefined,
417
+
hasSpaces: boolean,
418
+
hasFeeds: boolean,
419
+
): Promise<void> {
420
+
for (const op of MIGRATIONS) {
421
+
if (op.target === "spaces") {
422
+
if (!hasSpaces) continue;
423
+
await addColumnIfNotExists(spacesDb ?? db, op.table, op.column, op.columnDef);
424
+
continue;
268
425
}
426
+
if (op.target === "feeds") {
427
+
if (!hasFeeds) continue;
428
+
await addColumnIfNotExists(db, op.table, op.column, op.columnDef);
429
+
continue;
430
+
}
431
+
await addColumnIfNotExists(db, op.table, op.column, op.columnDef);
269
432
}
270
433
}
271
434
···
290
453
const base = buildSpacesBaseSchema(dialect);
291
454
const perCollection = buildCollectionTables(config, dialect, { forSpaces: true });
292
455
const indexes = buildDynamicIndexes(config, dialect, { forSpaces: true });
293
-
await target.batch([...base, ...perCollection, ...indexes].map((s) => target.prepare(s)));
456
+
// Per-statement (not batched) so concurrent applySpacesSchema on Postgres
457
+
// races only on the individual CREATE statements; see initSchema for
458
+
// rationale.
459
+
for (const stmt of [...base, ...perCollection, ...indexes]) {
460
+
await runIdempotentDdl(target, stmt);
461
+
}
294
462
295
463
const ftsStmts = buildFtsTables(config, dialect, { forSpaces: true });
296
464
for (const stmt of ftsStmts) {
297
465
try { await target.prepare(stmt).run(); } catch { /* FTS5 unavailable */ }
298
466
}
299
-
for (const stmt of buildCountColumns(config, { forSpaces: true })) {
300
-
try { await target.prepare(stmt).run(); } catch { /* already exists */ }
301
-
}
467
+
// Idempotent count-column ALTERs + their indexes. Non-duplicate-column
468
+
// errors propagate.
469
+
await applyCountColumns(target, config, { forSpaces: true });
302
470
}
303
471
304
472
export async function initSchema(
···
320
488
321
489
const all = [...baseStatements, ...collectionStatements, ...indexStatements, ...feedStatements];
322
490
323
-
await db.batch(all.map((s) => db.prepare(s)));
491
+
// Per-statement run (not a batched transaction) so concurrent initSchema
492
+
// callers on Postgres race only on individual CREATEs; the loser's
493
+
// duplicate-relation error is absorbed by runIdempotentDdl. Each statement
494
+
// is already idempotent (IF NOT EXISTS).
495
+
for (const stmt of all) {
496
+
await runIdempotentDdl(db, stmt);
497
+
}
324
498
325
499
if (config.spaces?.authority || config.spaces?.recordHost) {
326
500
await applySpacesSchema(spacesSharesMainDb ? db : spacesDb!, config, dialect);
···
339
513
// Labels tables live on the main DB — they're keyed by at-URI / DID and
340
514
// are read alongside public records during hydration.
341
515
const labelsStmts = buildLabelsSchema(dialect);
342
-
await db.batch(labelsStmts.map((s) => db.prepare(s)));
516
+
for (const stmt of labelsStmts) {
517
+
await runIdempotentDdl(db, stmt);
518
+
}
343
519
}
344
520
345
521
// FTS5 may not be available (e.g. node:sqlite) — skip gracefully
···
350
526
// FTS5 not supported in this environment
351
527
}
352
528
}
353
-
await runMigrations(db);
529
+
const hasSpaces = !!(config.spaces?.authority || config.spaces?.recordHost);
530
+
const hasFeeds = !!(config.feeds && Object.keys(config.feeds).length > 0);
531
+
// Spaces-targeted migrations route to spacesDb when one is configured;
532
+
// otherwise they hit the main db (which is where the spaces tables live
533
+
// when no separate spacesDb is supplied).
534
+
await runMigrations(db, spacesSharesMainDb ? undefined : spacesDb, hasSpaces, hasFeeds);
354
535
355
-
// Add count columns (ALTER TABLE — may already exist)
356
-
for (const stmt of buildCountColumns(config)) {
357
-
try {
358
-
await db.prepare(stmt).run();
359
-
} catch {
360
-
// Column/index already exists — ignore
361
-
}
362
-
}
536
+
// Idempotent count-column ALTERs + their indexes. Routed through
537
+
// `applyCountColumns` so non-duplicate-column errors propagate.
538
+
await applyCountColumns(db, config);
363
539
}
+45
-24
packages/contrail-appview/src/core/labels/resolve.ts
+45
-24
packages/contrail-appview/src/core/labels/resolve.ts
···
1
1
import {
2
2
CompositeDidDocumentResolver,
3
+
type DidDocumentResolver,
3
4
PlcDidDocumentResolver,
4
5
WebDidDocumentResolver,
5
6
} from "@atcute/identity-resolver";
6
7
import type { Did } from "@atcute/lexicons";
7
8
import type { Database } from "../types";
9
+
import { validateExternalUrl } from "../client";
8
10
9
-
/** Reject endpoint URLs that point to private/internal addresses or non-HTTPS.
10
-
* Mirrors the validator in core/client.ts — labeler endpoints should be
11
-
* publicly reachable for the same reasons PDS endpoints should. */
12
-
function validateEndpointUrl(url: string): boolean {
13
-
try {
14
-
const parsed = new URL(url);
15
-
if (parsed.protocol !== "https:") return false;
16
-
const host = parsed.hostname;
17
-
if (host === "localhost" || host === "127.0.0.1" || host === "[::1]") return false;
18
-
if (host.startsWith("10.")) return false;
19
-
if (host.startsWith("192.168.")) return false;
20
-
if (host.startsWith("169.254.")) return false;
21
-
if (/^172\.(1[6-9]|2\d|3[01])\./.test(host)) return false;
22
-
return true;
23
-
} catch {
24
-
return false;
25
-
}
11
+
/** Optional network-override knobs accepted by labeler-endpoint resolution.
12
+
* Mirrors the `ContrailConfig.networkOverrides` shape — kept narrow here so
13
+
* callers can pass `config.networkOverrides` directly without re-shaping.
14
+
* Omitting the object preserves the previous public-internet behavior. */
15
+
export interface LabelerResolveOverrides {
16
+
/** DID document resolver used when looking up the labeler service entry.
17
+
* When unset, falls back to a default composite (PLC + Web) pointing at the
18
+
* upstream PLC directory. Trusted; not SSRF-checked.
19
+
* Mirrors the resolver-injection pattern in `core/client.ts`. */
20
+
resolver?: DidDocumentResolver;
21
+
/** Hostnames (DNS names or IP literals) to allow past the default SSRF
22
+
* guard when validating a resolved labeler endpoint. Match is exact,
23
+
* case-insensitive, port-agnostic. */
24
+
additionalAllowedHosts?: string[];
26
25
}
27
26
28
-
const didResolver = new CompositeDidDocumentResolver({
27
+
/** Reject endpoint URLs that point to private/internal addresses or non-HTTPS.
28
+
* Thin alias for the single shared SSRF guard {@link validateExternalUrl} in
29
+
* `contrail-base` — labeler endpoints are validated by the exact same rules as
30
+
* PDS endpoints, so the allowlist logic must live in one place. Kept exported
31
+
* under this name for existing callers/tests. */
32
+
export const validateEndpointUrl = validateExternalUrl;
33
+
34
+
const DEFAULT_DID_RESOLVER: DidDocumentResolver = new CompositeDidDocumentResolver({
29
35
methods: {
30
36
plc: new PlcDidDocumentResolver(),
31
37
web: new WebDidDocumentResolver(),
···
33
39
});
34
40
35
41
/** Look up the labeler service endpoint from a DID.
36
-
* Reads the DID doc's `service[id="#atproto_labeler"].serviceEndpoint`. */
37
-
export async function resolveLabelerEndpoint(did: string): Promise<string | null> {
42
+
* Reads the DID doc's `service[id="#atproto_labeler"].serviceEndpoint`.
43
+
*
44
+
* `networkOverrides` (optional): customize the DID resolver used during the
45
+
* lookup, and/or which hostnames bypass the default SSRF guard. Omitting it
46
+
* preserves the original public-internet behavior. */
47
+
export async function resolveLabelerEndpoint(
48
+
did: string,
49
+
networkOverrides?: LabelerResolveOverrides,
50
+
): Promise<string | null> {
38
51
if (!did.startsWith("did:plc:") && !did.startsWith("did:web:")) return null;
52
+
const resolver = networkOverrides?.resolver ?? DEFAULT_DID_RESOLVER;
39
53
try {
40
-
const doc = await didResolver.resolve(did as Did<"plc"> | Did<"web">);
54
+
const doc = await resolver.resolve(did as Did<"plc"> | Did<"web">);
41
55
const endpoint = doc.service
42
56
?.find((s) => s.id === "#atproto_labeler")
43
57
?.serviceEndpoint?.toString();
44
58
if (!endpoint) return null;
45
-
if (!validateEndpointUrl(endpoint)) return null;
59
+
if (!validateEndpointUrl(endpoint, networkOverrides?.additionalAllowedHosts ?? [])) {
60
+
return null;
61
+
}
46
62
return endpoint;
47
63
} catch {
48
64
return null;
···
63
79
64
80
/** Get cached `(endpoint, cursor)` for a labeler. Resolves endpoint on
65
81
* cache miss or staleness; persists endpoint + resolved_at back to the DB
66
-
* so subsequent ingest cycles avoid the network round-trip. */
82
+
* so subsequent ingest cycles avoid the network round-trip.
83
+
*
84
+
* `networkOverrides` (optional): forwarded to `resolveLabelerEndpoint` for
85
+
* the cache-miss/stale path. Has no effect when `endpointOverride` is set
86
+
* or when a fresh cached endpoint is used. */
67
87
export async function getLabelerState(
68
88
db: Database,
69
89
did: string,
70
90
endpointOverride: string | undefined,
91
+
networkOverrides?: LabelerResolveOverrides,
71
92
): Promise<LabelerState | null> {
72
93
const row = await db
73
94
.prepare(
···
81
102
!row?.resolved_at || Date.now() - row.resolved_at > ENDPOINT_TTL_MS;
82
103
83
104
if (!endpoint || (!endpointOverride && stale)) {
84
-
endpoint = await resolveLabelerEndpoint(did);
105
+
endpoint = await resolveLabelerEndpoint(did, networkOverrides);
85
106
if (!endpoint) return null;
86
107
const now = Date.now();
87
108
await db
+26
-8
packages/contrail-appview/src/core/labels/subscribe.ts
+26
-8
packages/contrail-appview/src/core/labels/subscribe.ts
···
36
36
}
37
37
const remaining = Math.max(2_000, deadline - Date.now());
38
38
try {
39
-
await pumpOneLabeler(db, source, log, remaining, /* persistent */ false);
39
+
await pumpOneLabeler(
40
+
db,
41
+
source,
42
+
log,
43
+
remaining,
44
+
/* persistent */ false,
45
+
{},
46
+
config.networkOverrides,
47
+
);
40
48
} catch (err) {
41
49
log.warn(`[labels] cycle for ${source.did} failed: ${err}`);
42
50
}
···
62
70
const signal = options.signal;
63
71
64
72
const tasks = config.labels.sources.map((source) =>
65
-
runOneLabelerForever(db, source, log, signal, options),
73
+
runOneLabelerForever(db, source, log, signal, options, config.networkOverrides),
66
74
);
67
75
await Promise.all(tasks);
68
76
}
···
73
81
log: Logger,
74
82
signal: AbortSignal | undefined,
75
83
options: PersistentLabelsOptions,
84
+
networkOverrides: ContrailConfig["networkOverrides"],
76
85
): Promise<void> {
77
86
let attempts = 0;
78
87
while (!signal?.aborted) {
79
88
try {
80
-
await pumpOneLabeler(db, source, log, /* timeoutMs */ Infinity, true, {
81
-
signal,
82
-
batchSize: options.batchSize ?? DEFAULT_BATCH_SIZE,
83
-
flushIntervalMs: options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS,
84
-
});
89
+
await pumpOneLabeler(
90
+
db,
91
+
source,
92
+
log,
93
+
/* timeoutMs */ Infinity,
94
+
true,
95
+
{
96
+
signal,
97
+
batchSize: options.batchSize ?? DEFAULT_BATCH_SIZE,
98
+
flushIntervalMs: options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS,
99
+
},
100
+
networkOverrides,
101
+
);
85
102
attempts = 0;
86
103
} catch (err) {
87
104
if (signal?.aborted) break;
···
114
131
timeoutMs: number,
115
132
persistent: boolean,
116
133
pumpOpts: PumpOptions = {},
134
+
networkOverrides?: ContrailConfig["networkOverrides"],
117
135
): Promise<void> {
118
-
const state = await getLabelerState(db, source.did, source.endpoint);
136
+
const state = await getLabelerState(db, source.did, source.endpoint, networkOverrides);
119
137
if (!state) {
120
138
log.warn(`[labels] could not resolve labeler endpoint for ${source.did}; skipping`);
121
139
return;
+1
-1
packages/contrail-appview/src/core/router/collection.ts
+1
-1
packages/contrail-appview/src/core/router/collection.ts
···
301
301
302
302
let did: string | undefined;
303
303
if (actor) {
304
-
const resolved = await resolveActor(db, actor);
304
+
const resolved = await resolveActor(db, actor, config);
305
305
if (!resolved) throw new Error("Could not resolve actor");
306
306
did = resolved;
307
307
// backfillUser expects the record NSID (for PDS calls), not the short name.
+1
-1
packages/contrail-appview/src/core/router/feed.ts
+1
-1
packages/contrail-appview/src/core/router/feed.ts
···
243
243
return c.json({ error: "Unknown feed" }, 404);
244
244
}
245
245
246
-
const did = await resolveActor(db, actor);
246
+
const did = await resolveActor(db, actor, config);
247
247
if (!did) return c.json({ error: "Could not resolve actor" }, 400);
248
248
249
249
await maybeBackfillFeed(c, db, config, did, feedName, feedConfig);
+2
-2
packages/contrail-appview/src/core/router/index.ts
+2
-2
packages/contrail-appview/src/core/router/index.ts
···
88
88
const actor = c.req.query("actor");
89
89
if (!actor) return c.json({ error: "actor parameter required" }, 400);
90
90
91
-
const did = await resolveActor(db, actor);
91
+
const did = await resolveActor(db, actor, config);
92
92
if (!did) return c.json({ error: "Could not resolve actor" }, 400);
93
93
94
94
// Ensure profile records are backfilled
···
138
138
: config.spaces?.authority
139
139
? {
140
140
adapter: options.spaces?.adapter ?? new HostedAdapter(spacesDb, config),
141
-
verifier: buildVerifier(config.spaces.authority),
141
+
verifier: buildVerifier(config.spaces.authority, config.networkOverrides),
142
142
manifestVerifier: config.spaces.authority.signing
143
143
? createManifestVerifier({
144
144
resolveKey: async (iss) =>
+1
-1
packages/contrail-appview/src/core/router/notify.ts
+1
-1
packages/contrail-appview/src/core/router/notify.ts
+2
-2
packages/contrail-appview/src/core/router/profiles.ts
+2
-2
packages/contrail-appview/src/core/router/profiles.ts
···
85
85
}
86
86
87
87
// Resolve identities for all DIDs
88
-
const identities = await resolveIdentities(db, dids);
88
+
const identities = await resolveIdentities(db, dids, config);
89
89
90
90
// Fetch missing profile records from PDS on demand
91
91
const missingDids = dids.filter((d) => !result[d]);
···
134
134
const rkey = configRkey ?? "self";
135
135
const table = recordsTableName(shortName ?? collection);
136
136
try {
137
-
const pds = await getPDS(did as Did, db);
137
+
const pds = await getPDS(did as Did, db, config);
138
138
if (!pds) return;
139
139
140
140
const url = new URL("/xrpc/com.atproto.repo.getRecord", pds);
+1
-1
packages/contrail-appview/src/core/spaces/router.ts
+1
-1
packages/contrail-appview/src/core/spaces/router.ts
···
54
54
if (!authorityConfig) return;
55
55
56
56
const adapter = options.adapter ?? ctx?.adapter ?? new HostedAdapter(db, config);
57
-
const verifier = ctx?.verifier ?? buildVerifier(authorityConfig);
57
+
const verifier = ctx?.verifier ?? buildVerifier(authorityConfig, config.networkOverrides);
58
58
const auth = options.authMiddleware ?? createServiceAuthMiddleware(verifier);
59
59
60
60
const localRecordHost = spacesConfig.recordHost ? adapter : null;
+8
-7
packages/contrail-appview/src/core/spaces/schema.ts
+8
-7
packages/contrail-appview/src/core/spaces/schema.ts
···
5
5
buildCollectionTables,
6
6
buildDynamicIndexes,
7
7
buildFtsTables,
8
-
buildCountColumns,
8
+
applyCountColumns,
9
9
} from "../db/schema";
10
10
11
11
/** Spaces metadata tables — spaces, members, invites. No per-collection tables. */
···
76
76
77
77
/** Full spaces schema (base + per-collection tables + indexes). For callers
78
78
* that need a single array of statements. Note: this does NOT include FTS
79
-
* virtual tables or ALTER TABLE count columns — those must be applied with
80
-
* try/catch fallbacks and are handled by `initSchema`. */
79
+
* virtual tables or ALTER TABLE count columns — FTS is best-effort (engine
80
+
* may not be present) and count columns require dialect-aware idempotent
81
+
* ALTER. Both are handled by `initSchema` / `initSpacesSchema`. */
81
82
export function buildSpacesSchema(db: Database, config?: ContrailConfig): string[] {
82
83
const dialect = getDialect(db);
83
84
const base = buildSpacesBaseSchema(dialect);
···
94
95
const stmts = buildSpacesSchema(db, config);
95
96
await db.batch(stmts.map((s) => db.prepare(s)));
96
97
if (!config) return;
98
+
// FTS virtual tables: best-effort; the runtime may not have FTS5 compiled
99
+
// in (e.g. node:sqlite). Other DDL failures (count columns) propagate.
97
100
for (const stmt of buildFtsTables(config, dialect, { forSpaces: true })) {
98
-
try { await db.prepare(stmt).run(); } catch { /* ignore */ }
101
+
try { await db.prepare(stmt).run(); } catch { /* FTS5 unavailable */ }
99
102
}
100
-
for (const stmt of buildCountColumns(config, { forSpaces: true })) {
101
-
try { await db.prepare(stmt).run(); } catch { /* ignore */ }
102
-
}
103
+
await applyCountColumns(db, config, { forSpaces: true });
103
104
}