diff --git a/.env.example b/.env.example index a13fc37..445ed41 100644 --- a/.env.example +++ b/.env.example @@ -8,11 +8,16 @@ SIGNING_KEY=your-signing-key BSKY_IDENTIFIER=your-handle.bsky.social BSKY_PASSWORD=your-app-password -# Host to bind the labeler server to -# Use 0.0.0.0 in production (Railway) to bind to all interfaces -HOST=0.0.0.0 +# Public HTTP port for Caddy. Hosted platforms usually set PORT automatically. +PORT=8080 -# Port for the labeler server (AT Protocol labeler endpoint) +# Internal Next.js dashboard port behind Caddy +NEXT_PORT=3000 + +# Host to bind the labeler server to. 127.0.0.1 is enough when Caddy runs in the same container. +HOST=127.0.0.1 + +# Internal port for the labeler server (AT Protocol labeler endpoint behind Caddy) LABELER_PORT=4100 # Port for the Prometheus metrics endpoint @@ -24,21 +29,22 @@ ACTIVITY_DB_PATH=activity-log.db # Path to the labels database (used by @skyware/labeler LabelerServer) LABELS_DB_PATH=labels.db -# Public URL of the labeler endpoint (e.g. https://labeler.yourdomain.com) -# On Railway, set this to your service's public URL -LABELER_ENDPOINT=https://labeler.yourdomain.com +# Public URL of the Caddy-fronted app/labeler endpoint (e.g. https://labeler.yourdomain.com) +# /xrpc/* is routed to the labeler and dashboard/API routes are routed to Next.js. +NEXT_PUBLIC_LABELER_ENDPOINT=https://labeler.yourdomain.com # PDS URL for the labeler account (auto-detected during setup, override if needed) PDS_URL=https://climateai.org -# Tap sidecar configuration -TAP_URL=http://localhost:2480 +# External Tap service URL. Tap runs as a separate service, not inside this app container. +TAP_URL=https://your-tap-service.example.com TAP_ADMIN_PASSWORD= -TAP_BIND=:2480 -TAP_DB_PATH=tap.db + +# Comma-separated PDS hosts whose actors should always be labeled likely-test. +# When set, URL enrichment is deferred until each actor's PDS is known and skipped for matching test PDS hosts. +TEST_PDS_HOSTS= # --- Railway deployment --- # Mount a persistent volume at /app/data and set: # ACTIVITY_DB_PATH=/app/data/activity-log.db # LABELS_DB_PATH=/app/data/labels.db -# TAP_DB_PATH=/app/data/tap.db diff --git a/Caddyfile b/Caddyfile new file mode 100644 index 0000000..77e0710 --- /dev/null +++ b/Caddyfile @@ -0,0 +1,40 @@ +{ + auto_https off + admin off +} + +:{$PORT:8080} { + encode zstd gzip + + @xrpcPreflight { + path /xrpc/* + method OPTIONS + } + + handle @xrpcPreflight { + header Access-Control-Allow-Origin "*" + header Access-Control-Allow-Headers "authorization, content-type" + header Access-Control-Allow-Methods "GET, POST, OPTIONS" + header Access-Control-Max-Age "86400" + respond "" 204 + } + + @publicLabelerXrpc { + path /xrpc/_health /xrpc/com.atproto.label.queryLabels /xrpc/com.atproto.label.subscribeLabels + } + + handle @publicLabelerXrpc { + header { + Access-Control-Allow-Origin "*" + Cache-Control "no-store" + } + + reverse_proxy 127.0.0.1:{$LABELER_PORT:4100} { + flush_interval -1 + } + } + + handle { + reverse_proxy 127.0.0.1:{$NEXT_PORT:3000} + } +} diff --git a/DEPLOYMENT.md b/DEPLOYMENT.md index e0316c2..ce6e2a2 100644 --- a/DEPLOYMENT.md +++ b/DEPLOYMENT.md @@ -15,7 +15,9 @@ Current scripts: - `npm run start` → `next start` - `npm run dev:service` → runs `npm run dev` and `npm run labeler` together - `npm run labeler` → `tsx src/labeler/start.ts` -- `npm run start:service` → runs `npm run start` and `npm run labeler` together +- `npm run start:service` → runs Caddy, `npm run start:next`, and `npm run labeler` together +- `npm run start:next` → starts Next.js on `NEXT_PORT` (default `3000`) +- `npm run start:proxy` → starts Caddy using `Caddyfile` For hosted deployment, the image build uses `scripts/build.sh`, which sets `NEXT_PUBLIC_DEPLOY_TIME` and then runs `npm run build`. @@ -29,11 +31,13 @@ At runtime, the labeler process starts: Tap is a separate service. The labeler process connects to it via `TAP_URL` instead of launching it locally, and startup should fail if `TAP_URL` is missing. -The dashboard proxies only the public label distribution XRPC methods under `/xrpc/*` to the local labeler server at `http://127.0.0.1:LABELER_PORT`: +Caddy is the public front door for the app service. It routes dashboard and API traffic to Next.js, and routes public label distribution XRPC methods directly to the local labeler server at `http://127.0.0.1:LABELER_PORT`: - `com.atproto.label.queryLabels` - `com.atproto.label.subscribeLabels` +The direct Caddy route is required for `subscribeLabels`, because it is a WebSocket endpoint. + ## Environment variables ### Required for the running service @@ -72,8 +76,11 @@ If your Tap deployment uses admin auth, configure `TAP_ADMIN_PASSWORD` on the Ta - `HF_TOKEN` — enables HuggingFace classification; if unset, HF scoring stays disabled - `PDS_URL` — overrides PDS discovery from the DID document -- `HOST` — bind host for the labeler server (`0.0.0.0` by default) -- `LABELER_PORT` — labeler HTTP port (`4100` by default) +- `TEST_PDS_HOSTS` — comma-separated PDS hosts whose actors should always be labeled `likely-test`; URL enrichment waits for actor PDS resolution and skips matching test PDS hosts +- `PORT` — public Caddy HTTP port (hosted platforms usually set this; default fallback `8080`) +- `NEXT_PORT` — internal Next.js HTTP port (`3000` by default) +- `HOST` — bind host for the labeler server (`127.0.0.1` is recommended when Caddy runs in the same container) +- `LABELER_PORT` — internal labeler HTTP port (`4100` by default) - `METRICS_PORT` — metrics port (`4101` by default) - `RESET_DB=true` — deletes the SQLite files before startup @@ -103,6 +110,6 @@ Also persist the `-wal` and `-shm` companions that SQLite may create beside each ## Hosted environment notes - Run a single replica only for each service; SQLite state is not safe to share across multiple hosts. -- Expose the dashboard on the host-assigned HTTP port and make the labeler endpoint publicly reachable at `NEXT_PUBLIC_LABELER_ENDPOINT`. +- Expose Caddy on the host-assigned HTTP port and set `NEXT_PUBLIC_LABELER_ENDPOINT` to that public URL. Caddy will route `/xrpc/*` label methods to the labeler and dashboard/API routes to Next.js. - Keep Tap reachable from the app service at `TAP_URL`; do not rely on a localhost default. - If you need to reset state on a hosted platform, set `RESET_DB=true` for one restart, then remove it. diff --git a/Dockerfile b/Dockerfile index 5e4b4ae..314a303 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,8 +1,8 @@ # Build and run the app service FROM node:22-slim -# better-sqlite3 needs build tools -RUN apt-get update && apt-get install -y python3 make g++ && rm -rf /var/lib/apt/lists/* +# better-sqlite3 needs build tools; Caddy fronts Next and the labeler XRPC server +RUN apt-get update && apt-get install -y python3 make g++ caddy && rm -rf /var/lib/apt/lists/* WORKDIR /app @@ -20,8 +20,9 @@ ENV NEXT_PUBLIC_COMMIT_SHA=$RAILWAY_GIT_COMMIT_SHA ENV NEXT_PUBLIC_LABELER_ENDPOINT=$NEXT_PUBLIC_LABELER_ENDPOINT RUN chmod +x scripts/build.sh && ./scripts/build.sh -# Expose ports for the app service only; Tap runs separately via TAP_URL -EXPOSE 3000 4100 4101 +# Caddy listens on $PORT/8080; Next and labeler stay internal; Tap runs separately via TAP_URL +ENV NEXT_PORT=3000 +EXPOSE 8080 3000 4100 4101 # Run the app service (dashboard + labeler) CMD ["npm", "run", "start:service"] diff --git a/RAILWAY.md b/RAILWAY.md index afd68cf..76478e0 100644 --- a/RAILWAY.md +++ b/RAILWAY.md @@ -37,6 +37,7 @@ Required: | `SIGNING_KEY` | Labeler private key | | `NEXT_PUBLIC_LABELER_ENDPOINT` | Public HTTPS URL of the app service | | `TAP_URL` | Required URL of the separate Tap service; no localhost default | +| `TAP_ADMIN_PASSWORD` | App-side Tap admin password; required when the Tap service enables admin auth | Usually needed for setup/sync: @@ -57,10 +58,15 @@ Optional: | Variable | Description | Default | |----------|-------------|---------| | `HF_TOKEN` | Enables HuggingFace scoring | _(empty)_ | -| `HOST` | Labeler bind host | `0.0.0.0` | -| `LABELER_PORT` | Labeler HTTP port | `4100` | +| `NEXT_PORT` | Internal Next.js port behind Caddy | `3000` | +| `HOST` | Labeler bind host | `127.0.0.1` | +| `LABELER_PORT` | Internal labeler HTTP port behind Caddy | `4100` | | `METRICS_PORT` | Metrics port | `4101` | | `RESET_DB` | Clear app DB files on startup | _(empty)_ | +| `URL_ENRICHMENT_ENABLED` | Enable detachable async URL checks through `url_checks` | `true` | +| `URL_CHECK_TIMEOUT_MS` | Timeout for one URL resolution attempt | `4000` | +| `URL_CHECK_INTERVAL_MS` | URL worker polling interval | `1000` | +| `URL_CHECK_MAX_URLS_PER_DID` | Maximum profile/organization URLs cached and checked per DID | `5` | Start the app with: @@ -92,8 +98,9 @@ Do not point Tap at a custom database path unless the service itself exposes tha | Port | Service | Notes | |------|---------|-------| -| `$PORT` | App service | Railway routes web traffic here | -| `4100` | App service | Public labeler endpoint | +| `$PORT` | App service | Caddy listens here and routes traffic | +| `3000` | App service | Internal Next.js dashboard behind Caddy | +| `4100` | App service | Internal labeler endpoint behind Caddy | | `4101` | App service | Internal metrics only | | `2480` | Tap service | Tap HTTP endpoint | @@ -107,7 +114,7 @@ The app service must be able to reach `TAP_URL`; the Tap service does not need t 2. Set the app service `TAP_URL` to the Tap service URL. 3. Set the app service database paths to the mounted app volume. 4. Run `npm run setup` locally once to create the labeler credentials if needed. -5. Deploy the app service. +5. Deploy the app service. Caddy will route `/xrpc/com.atproto.label.queryLabels` and `/xrpc/com.atproto.label.subscribeLabels` directly to the labeler, and all dashboard/API routes to Next.js. --- @@ -127,7 +134,8 @@ The app service must be able to reach `TAP_URL`; the Tap service does not need t ### App service health check fails - Check `DID`, `SIGNING_KEY`, and `NEXT_PUBLIC_LABELER_ENDPOINT` -- Check the app logs for the labeler process error +- Check that Caddy is listening on `$PORT` +- Check the app logs for Next.js or labeler process errors ### Emergency reset diff --git a/README.md b/README.md index 12c0d86..7bcb2ea 100644 --- a/README.md +++ b/README.md @@ -43,12 +43,15 @@ The setup script will: ### Run ```bash -# Start the app service (dashboard + labeler) +# Start the app service locally (dashboard + labeler, without Caddy) npm run dev:service # Or start separately: npm run dev # Dashboard on http://localhost:3000 npm run labeler # Labeler backend on port 4100 + metrics on 4101 + +# Production start runs Caddy + Next + labeler +npm run start:service # Caddy on $PORT, Next on NEXT_PORT, labeler on LABELER_PORT ``` Tap runs as a separate service. Point `TAP_URL` at that service's URL; there is no localhost fallback and the app will not start without it. @@ -87,7 +90,13 @@ Scores `app.certified.actor.organization` records on 13 completeness signals (10 | Avatar | 10 | Has an avatar image | | Banner | 10 | Has a banner image | -Test detection: regex patterns catch common placeholder strings (`test`, `asdf`, `lorem ipsum`, etc.) and override the score to force ⚠ Likely Test. +Test detection: authenticity checks catch common placeholder strings (`test`, `asdf`, `lorem ipsum`, etc.) and override the score to force ⚠ Likely Test. Operators can also set `TEST_PDS_HOSTS` to force actors from known development PDS hosts into ⚠ Likely Test. Actor PDS lookup runs through the durable recompute queue; the first record from an uncached actor may be labeled by content score first, then corrected once the actor DID document is resolved. + +### URL enrichment + +Tap handlers never fetch URLs. New records are scored immediately with optimistic provisional URL resolve points for valid-looking public URLs. A detachable in-process URL enrichment worker checks those URLs later, stores results in the independent `url_checks` cache table, and queues a recompute only when cached URL state changes. + +When `TEST_PDS_HOSTS` is configured, URL enrichment is PDS-aware: it defers URL checks until the actor PDS cache is fresh, skips actors on configured test PDS hosts, and only checks URLs for actors on non-test PDS hosts. Set `URL_ENRICHMENT_ENABLED=false` to disable URL checks completely. When disabled, scoring keeps the provisional URL behavior and does not depend on the `url_checks` table. ## Scripts @@ -96,7 +105,7 @@ Test detection: regex patterns catch common placeholder strings (`test`, `asdf`, | `npm run dev` | Start Next.js dashboard | | `npm run labeler` | Start labeler backend | | `npm run dev:service` | Start dashboard + labeler concurrently | -| `npm run start:service` | Start the production app service processes | +| `npm run start:service` | Start Caddy reverse proxy + production dashboard + labeler process | | `npm run setup` | Initialize labeler account | | `npm run set-labels` | Push/update label definitions | | `npm run build` | Production build | @@ -112,32 +121,36 @@ Test detection: regex patterns catch common placeholder strings (`test`, `asdf`, | `BSKY_PASSWORD` | (set by setup) | Account password or app password | | `PDS_URL` | (auto-detected) | PDS endpoint URL | | `NEXT_PUBLIC_LABELER_ENDPOINT` | https://labeler. | Public HTTPS URL for the labeler | -| `HOST` | 127.0.0.1 | Labeler server bind address | -| `LABELER_PORT` | 4100 | Labeler server port | +| `PORT` | 8080 | Public HTTP port listened to by Caddy; hosted platforms usually set this | +| `NEXT_PORT` | 3000 | Internal Next.js port behind Caddy | +| `HOST` | 127.0.0.1 | Labeler server bind address; keep local when Caddy is in the same container | +| `LABELER_PORT` | 4100 | Internal labeler server port behind Caddy | | `METRICS_PORT` | 4101 | Prometheus metrics port | | `TAP_URL` | required | URL of the separate Tap service (no localhost default) | +| `TAP_ADMIN_PASSWORD` | empty | App-side password for Tap admin auth; must match the Tap service when auth is enabled | +| `TEST_PDS_HOSTS` | empty | Comma-separated PDS hosts whose actors should always be labeled `likely-test`; when set, URL enrichment waits for actor PDS resolution and skips matching test PDS hosts | | `ACTIVITY_DB_PATH` | `activity-log.db` | Activity log database path | +| `URL_ENRICHMENT_ENABLED` | `true` | Enables async URL checks through the detachable `url_checks` cache | +| `URL_CHECK_TIMEOUT_MS` | `4000` | Timeout for one URL resolution attempt | +| `URL_CHECK_INTERVAL_MS` | `1000` | Poll interval for the URL enrichment worker | +| `URL_CHECK_MAX_URLS_PER_DID` | `5` | Maximum profile/organization URLs cached and checked per DID | -Tap-specific settings belong on the Tap service, not the app service. This repo only needs `TAP_URL` here; set `TAP_ADMIN_PASSWORD` on the Tap service if you use Tap admin auth. +Tap runtime settings belong on the Tap service. If the Tap service sets `TAP_ADMIN_PASSWORD`, set the same value on the app service so health checks and the Tap WebSocket can authenticate. ## Production Deployment Deploy the app service and Tap as separate services. The app service runs the dashboard plus labeler backend and connects to Tap over `TAP_URL`; Tap owns its own database, volume, lifecycle, and any Tap-specific auth settings. -The labeler backend (port 4100) must be accessible via HTTPS for AT Protocol clients. Use a reverse proxy: - -```nginx -server { - listen 443 ssl; - server_name labeler.yourdomain.com; - - location / { - proxy_pass http://127.0.0.1:4100; - proxy_set_header Host $host; - } -} +The production app uses Caddy as the front door. Caddy routes public AT Protocol XRPC label methods directly to the labeler process and everything else to Next.js: + +```txt +/xrpc/com.atproto.label.queryLabels -> 127.0.0.1:4100 +/xrpc/com.atproto.label.subscribeLabels -> 127.0.0.1:4100 +/* -> 127.0.0.1:3000 ``` +This is important because `subscribeLabels` uses WebSockets, which need a real reverse proxy rather than the Next.js `fetch()` proxy fallback. + ## Tech Stack - **Runtime:** Node.js 22 diff --git a/docs/tap-worker-architecture.html b/docs/tap-worker-architecture.html new file mode 100644 index 0000000..75cb3cc --- /dev/null +++ b/docs/tap-worker-architecture.html @@ -0,0 +1,587 @@ + + + + + + Orglabeler Tap Worker Architecture + + + +
+
+
+

Tap worker architecture

+

Fast ingest, durable jobs, best-known labels.

+
+
+

Tap is only responsible for delivering events. The app persists snapshots, acks quickly, then workers score, enrich, classify, and relabel from local durable state.

+
+ No network I/O before ack + One org per DID: rkey self + Crash-safe job claims + Durable delete negation + Relabel only on tier changes +
+
+
+ +
+
+
+
+

Tap stream

+ external service +
+ +
+

Tap event

+

Profile or organization create/update/delete events arrive from Tap.

+
    +
  • app.certified.actor.profile
    • +
  • app.certified.actor.organization
    • +
  • org record invariant: rkey = self
    • +
+
+ +
handoff
+ +
+

Fast handler

+

Only does cheap local work.

+
    +
  • minimal parse/validation
    • +
  • persist latest snapshot/tombstone
    • +
  • upsert durable job
    • +
  • return so Tap can ack
    • +
+
+ +
+

Delete branch

+

Deletes do not negate labels inline. They persist a durable negate-org-label job with the record URI.

+
+
+ +
+
+

SQLite state

+ durable local source +
+ +
+

Snapshots

+

Latest actor state, keyed by DID.

+
    +
  • profile_snapshots[did]
    • +
  • organization_snapshots[did]
    • +
  • canonical org URI:
    at://{did}/app.certified.actor.organization/self
    • +
+
+ +
+

Typed jobs

+

One crash-safe queue with typed work.

+
+ recompute-org{did} + negate-org-label{did} + enrich-url{normalized_url} + classify-hf{did}:{model}:{hash} +
+
+ +
+

Worker claims

+

Jobs cannot be stranded by process restarts.

+
    +
  • pending → running → done
    • +
  • locked_until recovers stale running jobs
    • +
  • attempts + backoff + jitter
    • +
  • failed only after max attempts
    • +
+
+
+ +
+
+

Workers

+ after ack +
+ +
+

Debounced recompute

+

recompute-org:{did} waits 1–3 seconds so profile and org backfill can land together.

+
    +
  • reads latest snapshots
    • +
  • reads URL/HF caches
    • +
  • runs pure local scoring
    • +
  • applies best-known label quickly
    • +
+
+ +
+

URL enrichment

+

Network URL checks happen in a bounded worker, never inside Tap ack path.

+
    +
  • low concurrency
    • +
  • short timeout
    • +
  • cache by normalized URL
    • +
  • enqueue recompute after result
    • +
+
+ +
+

HF/BERT classification

+

Classifier worker writes cache rows and asks recompute to decide labels.

+
    +
  • key by model + input hash
    • +
  • pending has no effect
    • +
  • high-confidence low-quality can penalize/cap
    • +
  • recompute owns relabel decisions
    • +
+
+
+ +
+
+

Outputs

+ public + dashboard +
+ +
+

Pure scoring contract

+

Scoring uses local inputs only.

+
    +
  • profile + org snapshots
    • +
  • URL check cache
    • +
  • HF classification cache
    • +
  • returns score, tier, breakdown, enrichment status
    • +
+
+ +
+

Provisional label

+

Apply a label from best-known state immediately. The dashboard can mark enrichment as pending.

+
+ +
+

Later relabel

+

When URL/HF results arrive, recompute. If tier is unchanged, update the dashboard only. If tier changes, negate old quality label and apply the new one.

+
+ +
+

Delete correctness

+

negate-org-label retries remote label negation until success or max attempts. Local tombstone stays until public labels are cleaned up.

+
+
+
+
+ +
+
+ URL scoring +

Valid-looking URLs can receive provisional points. Confirmed failures can remove resolve points later.

+
+
+ HF scoring +

Pending classifier results have no effect. High-confidence low-quality results can add a penalty or tier cap.

+
+
+ DID keying +

Orglabeler has one organization record per DID, so recompute and org state are keyed by DID.

+
+
+ No churn +

Workers recompute often, but public label writes happen only when the effective tier changes.

+
+
+ +
+
+
0
+

Emergency fix

+

Keep URL network checks out of scoreActivity(); cap org URL checks.

+
+
+
1
+

Jobs + recompute

+

Add typed jobs, claim locks, debounce, and provisional best-known labels.

+
+
+
2
+

Durable deletes

+

Move label negation into retryable negate-org-label jobs.

+
+
+
3
+

URL + HF workers

+

Workers write enrichment caches and enqueue recompute; they never apply labels directly.

+
+
+
4
+

Observability

+

Track handler duration, queue depth, dead letters, and enrichment status in the dashboard.

+
+
+ + +
+ + diff --git a/docs/tap-worker-plan.md b/docs/tap-worker-plan.md new file mode 100644 index 0000000..3d43a7b --- /dev/null +++ b/docs/tap-worker-plan.md @@ -0,0 +1,205 @@ +# Tap Worker Plan + +## Goal + +Keep Tap ingestion fast and reliable by making the Tap handler do only cheap, durable local work. Anything that can block — URL checks, HuggingFace classification, label application retries, expensive recomputes — should run after Tap has been acknowledged. + +## Problem + +Tap only receives an ack after the event handler returns. When the handler performs network work, one slow URL or classifier request can hold the ack path long enough for Tap to retry events and grow the outbox buffer. + +Confirmed example: website resolution inside `scoreActivity()` blocked Tap acks. Removing network URL checks let the outbox drain. + +## Target flow + +```txt +Tap event + ↓ +Fast handler + - validate collection/action + - persist latest profile/org snapshot + - enqueue debounced recompute job for the DID + - return so Tap can ack + ↓ +Debounced recompute worker + - wait a short window so profile/org events can land together + - recompute from best-known local state + - apply/update a provisional label immediately + - update dashboard DB + ↓ +Enrichment workers + - resolve URLs with cache + timeout + - classify with HuggingFace/BERT + - enqueue recompute after results land + - relabel only if the effective tier changes +``` + +## Rules + +1. **No network I/O in the Tap handler** + - no URL fetches + - no HuggingFace calls + - no remote label reads/writes if avoidable + +2. **Tap handler must persist before returning** + - write snapshots/job rows to SQLite + - after that, the event can be safely acked + +3. **Jobs must be durable and coalesced** + - if profile and org events arrive close together, they should update one recompute job for the DID + - workers should process the latest state, not every intermediate event + +4. **Apply best-known labels immediately** + - do not wait for URL or HF enrichment before applying a label + - dashboard can show `pending enrichment`, but public labels should use the current best-known score + - later enrichment can trigger a recompute and relabel if the tier changes + +5. **Debounce actor recomputes briefly** + - profile and organization events for the same DID often arrive close together + - upsert one recompute job per DID with `run_after = now + 1–3 seconds` + - repeated events update the same job instead of producing multiple labels + +## Scoring model + +Scoring should be pure and local-only. It can read snapshots and cached enrichment rows, but it must never perform network I/O. + +Initial labels are **best-known provisional labels**. They are applied before URL checks or HuggingFace/BERT classification finish. + +### URL scoring before enrichment + +URL fields should be optimistic while enrichment is pending: + +| URL state | Scoring behavior | +| --- | --- | +| Missing URL | `0` URL points | +| Invalid URL syntax | `0` URL points | +| Valid-looking URL, not checked yet | award provisional URL points | +| Resolved OK later | keep/confirm URL points | +| Temporary check failure | keep provisional/unknown state and retry | +| Repeated hard failure | remove resolve points and enqueue recompute | + +This avoids delaying the first public label on slow or flaky URL checks. URL enrichment can still downgrade later if a URL repeatedly fails. + +### HF/BERT scoring before enrichment + +Classifier results should be conservative while pending: + +| HF/BERT state | Scoring behavior | +| --- | --- | +| Missing or pending result | no effect | +| Failed or timed out | no effect, retry later | +| Positive/meaningful result | no effect or small confirmation only | +| Low-quality result with high confidence | add penalty, authenticity signal, or tier cap | + +HF/BERT should not give optimistic points while pending. It should mainly act as a high-confidence negative signal because classifier false positives can otherwise cause label churn. + +### Recompute and relabel rule + +Any enrichment result writes to its cache table and then enqueues `recompute-org:{did}`. The recompute worker: + +1. reads latest profile/org snapshots, +2. reads cached URL and HF/BERT results, +3. computes the current best-known tier, +4. updates the dashboard row every time, and +5. writes public ATProto labels only when the effective tier changes. + +## Lightweight data model + +### `recompute_jobs` + +Actor-level work queue for scoring and label updates. `run_after` provides the short debounce window for profile/org events. + +```sql +id INTEGER PRIMARY KEY, +kind TEXT NOT NULL, -- recompute-org +key TEXT NOT NULL, -- DID or DID+rkey +status TEXT NOT NULL, -- pending, running, done, failed +attempts INTEGER NOT NULL, +run_after TEXT NOT NULL, +payload TEXT, +last_error TEXT, +created_at TEXT NOT NULL, +updated_at TEXT NOT NULL, +UNIQUE(kind, key) +``` + +### `url_checks` + +Cache URL resolution so the same URL is not fetched on every record update. + +```sql +normalized_url TEXT PRIMARY KEY, +status TEXT NOT NULL, -- pending, ok, failed +resolvable INTEGER, +status_code INTEGER, +error TEXT, +checked_at TEXT, +expires_at TEXT +``` + +### `hf_classifications` + +Stores classifier results keyed by model and input hash. + +```sql +did TEXT NOT NULL, +rkey TEXT NOT NULL, +model TEXT NOT NULL, +input_hash TEXT NOT NULL, +status TEXT NOT NULL, -- pending, done, failed +label TEXT, +score REAL, +error TEXT, +classified_at TEXT, +PRIMARY KEY (did, rkey, model, input_hash) +``` + +## Rollout phases + +### Phase 0 — current emergency fix + +- Keep URL network checks out of `scoreActivity()`. +- Keep org URL scoring bounded. +- Confirm Tap outbox drains after deploy. + +### Phase 1 — durable recompute worker + +Status: implemented. + +- `src/lib/db.ts` creates and manages `recompute_jobs` in the dashboard SQLite database. +- Tap handlers persist profile/org snapshots or pending delete rows, upsert one `recompute-org` job per DID, then return without scoring, HuggingFace calls, or label writes. +- Jobs use a 2 second debounce window so profile/org events can coalesce. +- `startRecomputeWorker()` drains due jobs, recomputes from best-known local state, writes dashboard rows, applies labels, and refreshes HuggingFace classification outside the Tap ack path. +- Organization deletes are persisted in `pending_organization_deletes` and negated/cleaned up by the worker or startup reconciliation. +- `/metrics` exposes `orglabeler_recompute_jobs{status="..."}` for durable job row status and `orglabeler_tap_handler_duration_ms` for Tap handler latency. + +### Phase 2 — URL enrichment worker + +Status: implemented. + +- `url_checks` is a detachable cache table. It stores only normalized URL resolution state and retry metadata; snapshots and activity rows do not depend on it. +- `src/labeler/url-enrichment-worker.ts` runs as an in-process polling worker with low concurrency, short fetch timeouts, manual redirect validation, bounded URLs per DID, and bounded retry/backoff. +- Scoring reads cached URL states as optional input. Missing or pending cache rows stay optimistic and keep provisional URL resolve points. +- `ok` URL checks confirm the provisional score. Repeated hard failures mark the URL `failed`, remove resolve points on the next recompute, and enqueue `recompute-org` for affected DIDs. +- The whole feature can be disabled with `URL_ENRICHMENT_ENABLED=false`; scoring then falls back to current provisional behavior. +- `/metrics` exposes `orglabeler_url_checks{status="..."}` for cache state counts. + +### Phase 3 — HuggingFace/BERT worker + +- Add `hf_classifications`. +- Hash classifier input so unchanged text reuses the same result. +- Process with low concurrency and retries/backoff. +- Recompute labels after classification lands. +- Relabel only if the effective tier changes. + +### Phase 4 — cleanup and observability + +- Show enrichment status in dashboard. +- Add dead-letter view for failed jobs. +- Add structured logs around job attempts, duration, and final label changes. + +## Open questions + +- What debounce window is enough for profile + organization backfill: 1s, 2s, or 3s? +- What retry policy should be used for URL checks and HF failures? +- How long should URL cache entries stay valid? diff --git a/docs/test-pds-option-2-design.html b/docs/test-pds-option-2-design.html new file mode 100644 index 0000000..2cbae07 --- /dev/null +++ b/docs/test-pds-option-2-design.html @@ -0,0 +1,1167 @@ + + + + + + TEST_PDS Option 2 — Orglabeler Design Doc + + + +
+
+
+

Orglabeler design doc · option 2

+

TEST_PDS through the existing recompute loop.

+
+ +
+ +
+ + +
+ + +
+
+ +
+
+

Cached test PDS path

+

The cached actor PDS host matches TEST_PDS_HOSTS, so the recompute result is rewritten to likely-test before DB logging and label sync.

+
+ override active +
+ +
+
+
Tap event writes snapshots1
+

Profile and organization records are persisted. The Tap handler still avoids slow external work.

+ snapshot saved +
+ +
+
Durable recompute job2
+

enqueueRecomputeJob('recompute-org', did) coalesces actor-level scoring work.

+ debounced +
+ +
+
Completeness scoring3
+

Build merged profile + organization input, score authenticity/completeness, and keep validation notes.

+ pure score +
+ +
+
Actor PDS cache check4
+

Read local actor_pds_cache. Missing or stale rows queue a PDS-resolution job.

+ cache hit +
+ +
+
PDS test override5
+

Add actor-pds-test-host: host to testSignals when the normalized host is configured. The numeric score stays unchanged.

+ tier forced to likely-test +
+ +
+
Log activity row6
+

Store score, tier, breakdown, test signals, validation notes, and HF fields as today.

+ DB source of truth +
+ +
+
Sync ATProto label7
+

Apply the DB tier to the organization record URI. Existing quality labels are negated if needed.

+ label applied +
+ +
+
Resolve DID → PDS8
+

For cache misses, the same worker handles resolve-actor-pds, fetches the DID document, and caches host state.

+ async only +
+ +
+
Correction recompute9
+

After resolving PDS, enqueue recompute-org again so DB, labels, and URL-enrichment eligibility converge on the cached host truth.

+ eventual correction +
+
+
+
+ +
+
+
+
+

Implementation seam

+

The key difference from hyperlabel: no standalone actor-PDS worker. Add one new recompute job kind and run it inside startRecomputeWorker().

+
// recompute worker loop
+if (job.kind === 'resolve-actor-pds') {
+  resolveAndCacheActorPds(job.key)
+  enqueueRecomputeJob('recompute-org', job.key)
+  completeRecomputeJob(job.id)
+}

+if (job.kind === 'recompute-org') {
+  await recomputeLabeledOrganizationRow(job.key)
+}
+
+ +
+

Why this shape

+

It preserves Tap ack speed, avoids a second interval worker, and lets the existing retry/backoff/recovery path handle PDS failures.

+
+
+ +
+

Files to touch

+
+
src/lib/config.tsAdd TEST_PDS_HOSTS parsed from env.
+
src/lib/pds-utils.tsNormalize hostnames, parse comma-separated values, exact-match configured hosts.
+
src/lib/pds-test-override.tsManage the stable test signal and derive the override tier.
+
src/lib/actor-pds-policy.tsCentralize cache-miss enqueueing and URL-enrichment gating.
+
src/lib/resolve-pds.tsAdd resolvePdsForDid() for did:plc and ideally did:web.
+
src/lib/db.tsAdd actor_pds_cache helpers and extend RecomputeJobKind.
+
src/labeler/tap-consumer.tsApply cached PDS override during recompute and process resolution jobs.
+
src/labeler/url-enrichment-worker.tsSkip URL checks for test-PDS actors and defer checks while PDS is unknown or stale.
+
README.md / DEPLOYMENT.mdDocument TEST_PDS_HOSTS, URL gating, and first-pass correction behavior.
+
+
+
+ +
+ + + + + + + + + + +
StepChangeDeveloper-readable behavior
1Parse envTEST_PDS_HOSTS="epds1.test.certified.app,dev.example" becomes normalized exact host list.
2Persist cacheactor_pds_cache stores pending | ok | failed, host, URL, expiry, and last error.
3Score normallyNormal authenticity/completeness score is produced first.
4Override tierIf cached host matches configured test hosts, add a test signal and force likely-test.
5Resolve laterMissing/stale cache rows queue resolve-actor-pds; resolved cache triggers another org recompute.
6Gate URL checksURL enrichment is deferred until PDS cache is fresh, skipped for configured test PDS hosts, and allowed for fresh non-test hosts.
+
+
+ +
+
+
+

State contracts

+
+
A
Config is optional.If TEST_PDS_HOSTS is empty, PDS cache checks and resolution jobs are skipped.
+
B
The score stays honest.PDS override does not mutate totalScore or completeness breakdown. It only changes testSignals and derived tier.
+
C
Signals are stable.Use a prefix like actor-pds-test-host: so stale PDS signals can be removed when config/cache changes.
+
D
Labels follow DB.The recompute result is logged first; then ATProto label sync uses the DB tier.
+
E
URL checks wait for PDS truth.With TEST_PDS_HOSTS set, URL enrichment only runs after a fresh non-test PDS cache hit.
+
+
+ +
+

Data model sketch

+
CREATE TABLE actor_pds_cache (
+  did TEXT PRIMARY KEY,
+  status TEXT NOT NULL CHECK(status IN ('pending', 'ok', 'failed')),
+  pds_url TEXT,
+  pds_host TEXT,
+  checked_at TEXT,
+  expires_at TEXT NOT NULL,
+  last_error TEXT,
+  created_at TEXT NOT NULL DEFAULT (datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
Exact host matching only. Start with normalized exact host equality. Wildcards look convenient, but they create surprising label scope.
+
+
+
+ +
+
+
+
+

Failure and edge-case behavior

+

The easy path should be correct, but the failure path must be boring: retry, keep prior data if safe, and tell operators what happened.

+
+ operator-visible +
+ +
+ + + + + + + + + + +
CaseBehaviorWhat to log / expose
Cache missQueue resolve-actor-pds; first recompute uses normal tier and URL enrichment is deferred.pending correction reason: cache miss.
Cache staleUse stale host if present for current recompute, refresh in background, and defer URL enrichment until fresh.refresh queued reason: stale cache.
DID document fetch failsRecord failed cache attempt and retry with bounded backoff.retry include HTTP status/error message.
Configured host removedRecompute removes stale actor-pds-test-host: signals and lets score tier win.self-healing count of rows updated.
Actor moves PDSTTL expiry refreshes cache; next recompute corrects labels and URL-enrichment eligibility.cache updated old host → new host.
Strict no-temporary-label requirementThis design is eventually correct. If temporary standard/high-quality labels are unacceptable, switch to blocking lookup or pre-seed cache.product decision document explicitly.
+
+
+
+
+
+ +
Self-contained design artifact: docs/test-pds-option-2-design.html
+
+ + + + diff --git a/package.json b/package.json index c817054..f87dbc4 100644 --- a/package.json +++ b/package.json @@ -7,9 +7,12 @@ "dev": "next dev", "build": "next build", "start": "next start", + "start:next": "next start -p ${NEXT_PORT:-3000}", + "start:proxy": "caddy run --config Caddyfile --adapter caddyfile", "labeler": "tsx src/labeler/start.ts", "dev:service": "concurrently --names \"next,labeler\" --prefix-colors \"blue,green\" \"npm run dev\" \"npm run labeler\"", - "start:service": "concurrently --names \"next,labeler\" --prefix-colors \"blue,green\" \"npm run start\" \"npm run labeler\"", + "start:service": "concurrently --kill-others-on-fail --names \"caddy,next,labeler\" --prefix-colors \"cyan,blue,green\" \"npm run start:proxy\" \"npm run start:next\" \"npm run labeler\"", + "test": "node --import tsx --test tests/**/*.test.ts", "set-labels": "tsx src/labeler/set-labels.ts", "setup": "tsx src/labeler/setup.ts", "reset": "tsx src/labeler/reset-db.ts", diff --git a/railway.toml b/railway.toml index a52e08d..1393fe5 100644 --- a/railway.toml +++ b/railway.toml @@ -9,6 +9,7 @@ watchPatterns = [ "package.json", "package-lock.json", "Dockerfile", + "Caddyfile", "railway.toml", "tsconfig.json", ] diff --git a/src/app/docs/page.tsx b/src/app/docs/page.tsx index 323b4cc..26d74d4 100644 --- a/src/app/docs/page.tsx +++ b/src/app/docs/page.tsx @@ -91,9 +91,9 @@ const API_ENDPOINTS = [ }, { method: 'GET', - path: '/xrpc/com.atproto.label.queryLabels?uriPatterns=at://did:plc:*/app.certified.actor.organization/*', - description: 'Query AT Protocol labels via the standard labeler endpoint. Use uriPatterns to filter record URIs and sources to scope the label source.', - curl: `curl "${labelerUrl('/xrpc/com.atproto.label.queryLabels?uriPatterns=at://did:plc:*/app.certified.actor.organization/*')}"`, + path: '/xrpc/com.atproto.label.queryLabels?uriPatterns=*', + description: 'Query AT Protocol labels via the standard labeler endpoint. Use uriPatterns with a full URI, a trailing-prefix wildcard, or * for all labels; sources scopes the label source.', + curl: `curl "${labelerUrl('/xrpc/com.atproto.label.queryLabels?uriPatterns=*')}"`, }, ] @@ -237,7 +237,9 @@ export default function DocsPage() {
diff --git a/src/labeler/metrics.ts b/src/labeler/metrics.ts index 1b4bd20..67775af 100644 --- a/src/labeler/metrics.ts +++ b/src/labeler/metrics.ts @@ -1,14 +1,55 @@ import express from 'express' -import { collectDefaultMetrics, register } from 'prom-client' +import { collectDefaultMetrics, Gauge, Histogram, register } from 'prom-client' +import { getRecomputeJobCounts, getUrlCheckCounts } from '../lib/db' import logger from './logger' collectDefaultMetrics() +const recomputeJobsGauge = new Gauge({ + name: 'orglabeler_recompute_jobs', + help: 'Durable recompute job rows grouped by status; done rows are retained coalescing state, not active queue depth', + labelNames: ['status'], +}) + +const tapHandlerDuration = new Histogram({ + name: 'orglabeler_tap_handler_duration_ms', + help: 'Tap handler wall-clock duration in milliseconds by collection and action', + labelNames: ['collection', 'action'], + buckets: [1, 5, 10, 25, 50, 100, 250, 500, 1000, 2500, 5000], +}) + +const urlChecksGauge = new Gauge({ + name: 'orglabeler_url_checks', + help: 'URL enrichment cache rows grouped by status', + labelNames: ['status'], +}) + +/** Records how long the Tap handler took before the event could be acknowledged. */ +export function observeTapHandlerDuration(collection: string, action: string, durationMs: number): void { + tapHandlerDuration.observe({ collection, action }, durationMs) +} + +function updateRecomputeJobMetrics(): void { + const counts = getRecomputeJobCounts() + for (const [status, count] of Object.entries(counts)) { + recomputeJobsGauge.set({ status }, count) + } +} + +function updateUrlCheckMetrics(): void { + const counts = getUrlCheckCounts() + for (const [status, count] of Object.entries(counts)) { + urlChecksGauge.set({ status }, count) + } +} + export function startMetricsServer(port: number): ReturnType { const app = express() app.get('/metrics', async (_req, res) => { try { + updateRecomputeJobMetrics() + updateUrlCheckMetrics() res.set('Content-Type', register.contentType) res.end(await register.metrics()) } catch (err) { diff --git a/src/labeler/start.ts b/src/labeler/start.ts index b277627..4c73c0d 100644 --- a/src/labeler/start.ts +++ b/src/labeler/start.ts @@ -8,9 +8,11 @@ import { backfillHfClassification, syncLabelsWithDb, reconcileStoredOrganizationSnapshots, + startRecomputeWorker, } from './tap-consumer' import { setReclassifyCallback } from '../lib/hf-classifier' import { startMetricsServer } from './metrics' +import { startUrlEnrichmentWorker } from './url-enrichment-worker' import logger from './logger' // Fix 3 & 4: module-scope shuttingDown flag used by shutdown @@ -24,15 +26,26 @@ async function waitForTap(url: string, maxAttempts = 30, intervalMs = 1000): Pro : undefined const res = await fetch(`${url}/health`, authHeader ? { headers: authHeader } : undefined) if (TAP_ADMIN_PASSWORD) { + if (res.status === 401 || res.status === 403) { + throw new Error('Tap rejected TAP_ADMIN_PASSWORD; check that the app and Tap service use the same admin password') + } if (res.ok) { logger.info({ status: res.status }, 'Tap is healthy') return } - } else if (res.status < 500) { - logger.info({ status: res.status }, 'Tap is healthy') - return + } else { + if (res.status === 401 || res.status === 403) { + throw new Error('Tap requires admin auth. Set TAP_ADMIN_PASSWORD on the app service to match the Tap service.') + } + if (res.status < 500) { + logger.info({ status: res.status }, 'Tap is healthy') + return + } + } + } catch (err) { + if (err instanceof Error && err.message.includes('TAP_ADMIN_PASSWORD')) { + throw err } - } catch { // tap not ready yet — connection refused or similar } await new Promise(r => setTimeout(r, intervalMs)) @@ -44,12 +57,16 @@ async function main() { // Fix 4: declare consumer at outer scope so shutdown can access it // even if a signal arrives during startup let consumer: Awaited> | undefined + let recomputeWorker: ReturnType | undefined + let urlEnrichmentWorker: ReturnType | undefined // Fix 4: register shutdown handlers EARLY, before any async work async function shutdown(signal: string) { if (shuttingDown) return shuttingDown = true logger.info({ signal }, 'Shutting down...') + urlEnrichmentWorker?.destroy() + recomputeWorker?.destroy() await consumer?.destroy() await new Promise((resolve) => { labelerServer.close(() => resolve()) @@ -123,7 +140,11 @@ async function main() { // 10. Wait for tap to be ready await waitForTap(TAP_URL) - // 11. Start tap consumer (replaces Jetstream subscription) + // 11. Start async workers before Tap so persisted jobs can drain. + recomputeWorker = startRecomputeWorker() + urlEnrichmentWorker = startUrlEnrichmentWorker() + + // 12. Start tap consumer (replaces Jetstream subscription) consumer = startTapConsumer() logger.info('Tap consumer started — receiving backfill + live events') backfillHfClassification() diff --git a/src/labeler/tap-consumer.ts b/src/labeler/tap-consumer.ts index ffad2e9..f292f00 100644 --- a/src/labeler/tap-consumer.ts +++ b/src/labeler/tap-consumer.ts @@ -1,19 +1,33 @@ import { Tap, SimpleIndexer } from '@atproto/tap' import type { TapChannel } from '@atproto/tap' import { TAP_URL, TAP_ADMIN_PASSWORD, ACTIVITY_COLLECTION } from '../lib/config' +import { cachedActorPdsHostForScoring, ACTOR_PDS_CACHE_TTL_MS } from '../lib/actor-pds-policy' +import { applyPdsTestOverride } from '../lib/pds-test-override' +import { normalizePdsHostFromUrl } from '../lib/pds-utils' +import { resolvePdsForDid } from '../lib/resolve-pds' import { scoreActivity } from '../lib/scorer' import { logActivity, getUnclassifiedActivities, getAllActivitiesForSync, + claimDueRecomputeJob, + completeRecomputeJob, + enqueueRecomputeJob, + failRecomputeJob, + getRecomputeJobCounts, + hasPendingOrganizationDelete, + recoverStaleRunningRecomputeJobs, getProfileSnapshot, getOrganizationSnapshot, getAllOrganizationSnapshots, deleteProfileSnapshot, deleteOrganizationRecordState, + completePendingOrganizationDelete, deletePendingOrganizationDelete, clearActivityHfFields, getPendingOrganizationDeletes, + recordActorPdsFailure, + recordActorPdsOk, recordPendingOrganizationDeleteAttempt, upsertProfileSnapshot, upsertOrganizationSnapshot, @@ -25,6 +39,8 @@ import { getMergedActorDisplay } from '../lib/actor-display' import { buildMergedScoringInput } from '../lib/scoring-input' import { salvageProfileFallback } from '../lib/profile-fallback' import logger from './logger' +import { observeTapHandlerDuration } from './metrics' +import { enqueueUrlChecksForDid, getUrlResolutionMapForDid } from './url-enrichment-worker' import type { ActivityRecord, ProfileFallbackProfile, ProfileSnapshot, RuntimeLabelTier } from '../lib/types' import { $safeParse as $safeParseProfile } from '../lexicons/app/certified/actor/profile.defs' import { $safeParse } from '../lexicons/app/certified/actor/organization.defs' @@ -35,6 +51,13 @@ const tap = new Tap(TAP_URL, tapConfig) const PROFILE_COLLECTION = 'app.certified.actor.profile' const ORGANIZATION_COLLECTION = ACTIVITY_COLLECTION const TARGET_COLLECTIONS = [PROFILE_COLLECTION, ORGANIZATION_COLLECTION] +const RECOMPUTE_DEBOUNCE_MS = 2000 +const RECOMPUTE_WORKER_INTERVAL_MS = 500 +const RECOMPUTE_MAX_RETRY_DELAY_MS = 60_000 +const RECOMPUTE_MAX_ATTEMPTS = 10 +const RECOMPUTE_STALE_RUNNING_AFTER_MS = 5 * 60_000 +const PENDING_DELETE_RETRY_BASE_MS = 2000 +const PENDING_DELETE_MAX_RETRY_MS = 60_000 const indexer = new SimpleIndexer() @@ -90,7 +113,7 @@ function logRecordOutcome(details: { did: string collection: string action: string - source: 'live' | 'startup' + source: 'live' | 'startup' | 'worker' labelAction: string score?: number tier?: RuntimeLabelTier @@ -99,6 +122,34 @@ function logRecordOutcome(details: { logger.info(details, 'Processed Tap record') } +function enqueueOrganizationRecompute(did: string, reason: string): void { + enqueueRecomputeJob('recompute-org', did, { + delayMs: RECOMPUTE_DEBOUNCE_MS, + payload: { reason }, + }) +} + +function retryDelayForAttempt(attempts: number): number { + const exponent = Math.max(0, attempts - 1) + return Math.min(RECOMPUTE_MAX_RETRY_DELAY_MS, RECOMPUTE_DEBOUNCE_MS * (2 ** exponent)) +} + +function pendingDeleteRetryDelayForAttempt(attempts: number): number { + const exponent = Math.max(0, attempts - 1) + return Math.min(PENDING_DELETE_MAX_RETRY_MS, PENDING_DELETE_RETRY_BASE_MS * (2 ** exponent)) +} + +async function resolveAndCacheActorPds(did: string): Promise<{ pdsUrl: string; pdsHost: string }> { + const { pds } = await resolvePdsForDid(did) + const pdsHost = normalizePdsHostFromUrl(pds) + if (!pdsHost) { + throw new Error(`Resolved PDS endpoint for ${did} is not a valid URL: ${pds}`) + } + + recordActorPdsOk(did, pds, pdsHost, ACTOR_PDS_CACHE_TTL_MS) + return { pdsUrl: pds, pdsHost } +} + function buildHfText( profileSnapshot: ReturnType, organization: ActivityRecord, @@ -148,6 +199,8 @@ type OrganizationRecomputeOutcome = { } export async function recomputeLabeledOrganizationRow(did: string): Promise { + if (hasPendingOrganizationDelete(did)) return null + const organization = getOrganizationSnapshot(did) if (!organization) return null @@ -158,6 +211,8 @@ export async function recomputeLabeledOrganizationRow(did: string): Promise { - const snapshots = getAllOrganizationSnapshots() - - for (const snapshot of snapshots) { - const outcome = await recomputeLabeledOrganizationRow(snapshot.did) - if (outcome) { - logRecordOutcome({ - did: snapshot.did, - collection: ORGANIZATION_COLLECTION, - action: 'reconcile', - source: 'startup', - score: outcome.score, - tier: outcome.tier, - labelAction: outcome.labelAction, - profileIngestMode: outcome.profileIngestMode, - }) - } - } - +async function processPendingOrganizationDeletes(source: 'startup' | 'worker'): Promise { const pendingDeletes = getPendingOrganizationDeletes() + let processed = 0 + for (const pendingDelete of pendingDeletes) { try { logger.info( - { did: pendingDelete.did, rkey: pendingDelete.rkey, uri: pendingDelete.record_uri }, + { did: pendingDelete.did, rkey: pendingDelete.rkey, uri: pendingDelete.record_uri, source }, 'Retrying pending organization label negation before cleanup', ) - await negateQualityLabels(pendingDelete.record_uri) - deleteOrganizationRecordState(pendingDelete.did, pendingDelete.rkey) - deletePendingOrganizationDelete(pendingDelete.did) + const negatedCount = await negateQualityLabels(pendingDelete.record_uri) + const completed = completePendingOrganizationDelete( + pendingDelete.did, + pendingDelete.rkey, + pendingDelete.record_uri, + ) + + if (!completed) { + logger.info( + { did: pendingDelete.did, rkey: pendingDelete.rkey, uri: pendingDelete.record_uri, source }, + 'Pending organization delete was already superseded; skipped local cleanup', + ) + continue + } + + processed++ + + logRecordOutcome({ + did: pendingDelete.did, + collection: ORGANIZATION_COLLECTION, + action: 'delete', + source, + labelAction: negatedCount > 0 ? 'negated' : 'unchanged', + profileIngestMode: getProfileIngestMode(getProfileSnapshot(pendingDelete.did)), + }) } catch (err) { const message = err instanceof Error ? err.message : String(err) - recordPendingOrganizationDeleteAttempt(pendingDelete.did, message) + const attempts = pendingDelete.attempts + 1 + const retryDelayMs = pendingDeleteRetryDelayForAttempt(attempts) + recordPendingOrganizationDeleteAttempt(pendingDelete.did, message, retryDelayMs) logger.error( - { err, did: pendingDelete.did, rkey: pendingDelete.rkey, uri: pendingDelete.record_uri }, - 'Pending organization delete negation failed; will retry later', + { err, did: pendingDelete.did, rkey: pendingDelete.rkey, uri: pendingDelete.record_uri, source, retryDelayMs }, + 'Pending organization delete negation failed; retry scheduled', ) } } + return processed +} + +export async function reconcileStoredOrganizationSnapshots(): Promise { + await processPendingOrganizationDeletes('startup') + + const snapshots = getAllOrganizationSnapshots() + + for (const snapshot of snapshots) { + try { + const outcome = await recomputeLabeledOrganizationRow(snapshot.did) + if (outcome) { + logRecordOutcome({ + did: snapshot.did, + collection: ORGANIZATION_COLLECTION, + action: 'reconcile', + source: 'startup', + score: outcome.score, + tier: outcome.tier, + labelAction: outcome.labelAction, + profileIngestMode: outcome.profileIngestMode, + }) + } + } catch (err) { + logger.error({ err, did: snapshot.did, rkey: snapshot.rkey }, 'Startup reconciliation failed; queued durable recompute retry') + enqueueOrganizationRecompute(snapshot.did, 'startup-reconcile-failed') + } + } + if (snapshots.length > 0) { logger.info({ count: snapshots.length }, 'Reconciled local organization snapshots on startup') } else { @@ -273,199 +364,163 @@ async function handleOrganizationDelete(did: string, rkey: string): Promise 0 ? 'negated' : 'unchanged', - profileIngestMode: getProfileIngestMode(getProfileSnapshot(did)), - }) - } catch (err) { - const message = err instanceof Error ? err.message : String(err) - recordPendingOrganizationDeleteAttempt(did, message) - logger.error({ err, did, rkey, uri: recordUri }, 'Error negating organization labels after delete') - } + logRecordOutcome({ + did, + collection: ORGANIZATION_COLLECTION, + action: 'delete', + source: 'live', + labelAction: 'delete-queued', + profileIngestMode: getProfileIngestMode(getProfileSnapshot(did)), + }) } async function handleProfileDelete(did: string): Promise { deleteProfileSnapshot(did) if (getOrganizationSnapshot(did)) { - const outcome = await recomputeLabeledOrganizationRow(did) - refreshHfClassification(did) - - if (outcome) { - logRecordOutcome({ - did, - collection: PROFILE_COLLECTION, - action: 'delete', - source: 'live', - score: outcome.score, - tier: outcome.tier, - labelAction: outcome.labelAction, - profileIngestMode: outcome.profileIngestMode, - }) - } - } else { - logRecordOutcome({ - did, - collection: PROFILE_COLLECTION, - action: 'delete', - source: 'live', - labelAction: 'profile-deleted', - profileIngestMode: 'missing', - }) + enqueueOrganizationRecompute(did, 'profile-delete') } + + logRecordOutcome({ + did, + collection: PROFILE_COLLECTION, + action: 'delete', + source: 'live', + labelAction: getOrganizationSnapshot(did) ? 'recompute-queued' : 'profile-deleted', + profileIngestMode: 'missing', + }) } indexer.record(async (evt) => { - const eventMeta = { - collection: evt.collection, - action: evt.action, - did: evt.did, - rkey: evt.rkey, - } - - if (evt.collection === PROFILE_COLLECTION) { - if (evt.action === 'delete') { - await handleProfileDelete(evt.did) - return - } - - if (!evt.record) { - logger.warn(eventMeta, 'Skipping profile event with missing record payload') - return + const startedAt = performance.now() + try { + const eventMeta = { + collection: evt.collection, + action: evt.action, + did: evt.did, + rkey: evt.rkey, } - const parsed = $safeParseProfile(evt.record) - if (parsed.success) { - upsertProfileSnapshot({ - did: evt.did, - recordUri: `at://${evt.did}/${PROFILE_COLLECTION}/${evt.rkey}`, - rkey: evt.rkey, - payload: parsed.value, - validationNotes: [], - updatedAt: new Date().toISOString(), - }) - } else { - const fallback = salvageProfileFallback(evt.record) - const preserved = fallback.mode === 'fallback' - ? summarizeFallbackPreservation(fallback.profile) - : summarizeFallbackPreservation(null) - - logger.warn( - { ...eventMeta, reason: parsed.reason?.message }, - 'Profile record failed strict lexicon validation; attempting fallback', - ) - - if (fallback.mode === 'unusable') { - const { noteCount, noteSummary } = summarizeValidationNotes(fallback.validationNotes) + if (evt.collection === PROFILE_COLLECTION) { + if (evt.action === 'delete') { + await handleProfileDelete(evt.did) + return + } - logger.warn( - { - ...eventMeta, - fallbackSucceeded: false, - reason: parsed.reason?.message, - noteCount, - noteSummary, - fallbackReason: fallback.validationNotes[0], - preserved, - }, - 'Profile record was unusable after fallback; skipping', - ) + if (!evt.record) { + logger.warn(eventMeta, 'Skipping profile event with missing record payload') return } - upsertProfileSnapshot({ - did: evt.did, - recordUri: `at://${evt.did}/${PROFILE_COLLECTION}/${evt.rkey}`, - rkey: evt.rkey, - payload: fallback.profile as ProfileSnapshot['payload'], - validationNotes: fallback.validationNotes, - updatedAt: new Date().toISOString(), - }) - } + const parsed = $safeParseProfile(evt.record) + if (parsed.success) { + upsertProfileSnapshot({ + did: evt.did, + recordUri: `at://${evt.did}/${PROFILE_COLLECTION}/${evt.rkey}`, + rkey: evt.rkey, + payload: parsed.value, + validationNotes: [], + updatedAt: new Date().toISOString(), + }) + } else { + const fallback = salvageProfileFallback(evt.record) + const preserved = fallback.mode === 'fallback' + ? summarizeFallbackPreservation(fallback.profile) + : summarizeFallbackPreservation(null) - if (getOrganizationSnapshot(evt.did)) { - const outcome = await recomputeLabeledOrganizationRow(evt.did) - refreshHfClassification(evt.did) + logger.warn( + { ...eventMeta, reason: parsed.reason?.message }, + 'Profile record failed strict lexicon validation; attempting fallback', + ) - if (outcome) { - logRecordOutcome({ + if (fallback.mode === 'unusable') { + const { noteCount, noteSummary } = summarizeValidationNotes(fallback.validationNotes) + + logger.warn( + { + ...eventMeta, + fallbackSucceeded: false, + reason: parsed.reason?.message, + noteCount, + noteSummary, + fallbackReason: fallback.validationNotes[0], + preserved, + }, + 'Profile record was unusable after fallback; skipping', + ) + return + } + + upsertProfileSnapshot({ did: evt.did, - collection: PROFILE_COLLECTION, - action: evt.action, - source: 'live', - score: outcome.score, - tier: outcome.tier, - labelAction: outcome.labelAction, - profileIngestMode: getProfileIngestMode(getProfileSnapshot(evt.did)), + recordUri: `at://${evt.did}/${PROFILE_COLLECTION}/${evt.rkey}`, + rkey: evt.rkey, + payload: fallback.profile as ProfileSnapshot['payload'], + validationNotes: fallback.validationNotes, + updatedAt: new Date().toISOString(), }) } - } else { + + if (getOrganizationSnapshot(evt.did)) { + enqueueOrganizationRecompute(evt.did, 'profile-upsert') + } + logRecordOutcome({ did: evt.did, collection: PROFILE_COLLECTION, action: evt.action, source: 'live', - labelAction: 'profile-saved', + labelAction: getOrganizationSnapshot(evt.did) ? 'recompute-queued' : 'profile-saved', profileIngestMode: getProfileIngestMode(getProfileSnapshot(evt.did)), }) - } - return - } + return + } - if (evt.collection !== ORGANIZATION_COLLECTION) { - return - } + if (evt.collection !== ORGANIZATION_COLLECTION) { + return + } - if (evt.action === 'delete') { - await handleOrganizationDelete(evt.did, evt.rkey) - return - } + if (evt.action === 'delete') { + await handleOrganizationDelete(evt.did, evt.rkey) + return + } - if (!evt.record) { - logger.warn(eventMeta, 'Skipping organization event with missing record payload') - return - } + if (!evt.record) { + logger.warn(eventMeta, 'Skipping organization event with missing record payload') + return + } - // Validate against the app.certified.actor.organization lexicon - const parsed = $safeParse(evt.record) - if (!parsed.success) { - logger.warn({ ...eventMeta, reason: parsed.reason?.message }, 'Organization record failed lexicon validation — skipping') - return - } - const record = parsed.value as ActivityRecord - upsertOrganizationSnapshot({ - did: evt.did, - recordUri: `at://${evt.did}/${ORGANIZATION_COLLECTION}/${evt.rkey}`, - rkey: evt.rkey, - payload: record, - updatedAt: new Date().toISOString(), - }) + // Validate against the app.certified.actor.organization lexicon + const parsed = $safeParse(evt.record) + if (!parsed.success) { + logger.warn({ ...eventMeta, reason: parsed.reason?.message }, 'Organization record failed lexicon validation — skipping') + return + } + const record = parsed.value as ActivityRecord + deletePendingOrganizationDelete(evt.did) + upsertOrganizationSnapshot({ + did: evt.did, + recordUri: `at://${evt.did}/${ORGANIZATION_COLLECTION}/${evt.rkey}`, + rkey: evt.rkey, + payload: record, + updatedAt: new Date().toISOString(), + }) - const outcome = await recomputeLabeledOrganizationRow(evt.did) - refreshHfClassification(evt.did) + enqueueOrganizationRecompute(evt.did, 'organization-upsert') - if (outcome) { logRecordOutcome({ did: evt.did, collection: ORGANIZATION_COLLECTION, action: evt.action, source: 'live', - score: outcome.score, - tier: outcome.tier, - labelAction: outcome.labelAction, - profileIngestMode: outcome.profileIngestMode, + labelAction: 'recompute-queued', + profileIngestMode: getProfileIngestMode(getProfileSnapshot(evt.did)), }) + } finally { + observeTapHandlerDuration(evt.collection, evt.action, performance.now() - startedAt) } }) @@ -527,6 +582,107 @@ export async function syncLabelsWithDb(): Promise { } } +/** + * Starts the durable recompute loop that drains debounced actor jobs after Tap + * events have been acknowledged. The worker performs scoring, actor PDS resolution, + * label writes, and pending delete cleanup outside the Tap handler. + */ +export function startRecomputeWorker(): { destroy: () => void } { + let stopped = false + let running = false + let timer: NodeJS.Timeout | null = null + + const tick = async (): Promise => { + if (stopped || running) return + running = true + + try { + const recovered = recoverStaleRunningRecomputeJobs(RECOMPUTE_STALE_RUNNING_AFTER_MS) + if (recovered > 0) { + logger.warn({ recovered }, 'Recovered stale running recompute jobs') + } + + await processPendingOrganizationDeletes('worker') + + let job = claimDueRecomputeJob() + while (job) { + const currentJob = job + try { + if (currentJob.kind === 'resolve-actor-pds') { + const resolved = await resolveAndCacheActorPds(currentJob.key) + enqueueOrganizationRecompute(currentJob.key, 'actor-pds-resolved') + completeRecomputeJob(currentJob.id) + logger.info( + { did: currentJob.key, pdsUrl: resolved.pdsUrl, pdsHost: resolved.pdsHost }, + 'Resolved actor PDS and queued organization recompute', + ) + job = claimDueRecomputeJob() + continue + } + + if (currentJob.kind !== 'recompute-org') { + completeRecomputeJob(currentJob.id) + job = claimDueRecomputeJob() + continue + } + + const outcome = await recomputeLabeledOrganizationRow(currentJob.key) + if (outcome) { + enqueueUrlChecksForDid(currentJob.key) + refreshHfClassification(currentJob.key) + logRecordOutcome({ + did: currentJob.key, + collection: ORGANIZATION_COLLECTION, + action: 'recompute', + source: 'worker', + score: outcome.score, + tier: outcome.tier, + labelAction: outcome.labelAction, + profileIngestMode: outcome.profileIngestMode, + }) + } + + completeRecomputeJob(currentJob.id) + } catch (err) { + const message = err instanceof Error ? err.message : String(err) + const retryDelayMs = retryDelayForAttempt(currentJob.attempts) + + if (currentJob.kind === 'resolve-actor-pds') { + recordActorPdsFailure(currentJob.key, message, retryDelayMs) + } + + if (currentJob.attempts >= RECOMPUTE_MAX_ATTEMPTS) { + logger.error({ err, job: currentJob }, 'Recompute job exceeded max attempts; leaving failed for inspection') + failRecomputeJob(currentJob.id, message, RECOMPUTE_MAX_RETRY_DELAY_MS, 'failed') + } else { + failRecomputeJob(currentJob.id, message, retryDelayMs) + logger.error({ err, job: currentJob, retryDelayMs }, 'Recompute job failed; retry scheduled') + } + } + + job = claimDueRecomputeJob() + } + } finally { + running = false + } + } + + timer = setInterval(() => { + tick().catch(err => logger.error({ err }, 'Recompute worker tick failed')) + }, RECOMPUTE_WORKER_INTERVAL_MS) + + logger.info({ counts: getRecomputeJobCounts() }, 'Recompute worker started') + tick().catch(err => logger.error({ err }, 'Initial recompute worker tick failed')) + + return { + destroy: () => { + stopped = true + if (timer) clearInterval(timer) + logger.info({ counts: getRecomputeJobCounts() }, 'Recompute worker stopped') + }, + } +} + export function startTapConsumer(): { channel: TapChannel; destroy: () => Promise } { logger.info({ collections: TARGET_COLLECTIONS }, 'Starting Tap consumer for target collections') const channel = tap.channel(indexer) diff --git a/src/labeler/url-enrichment-worker.ts b/src/labeler/url-enrichment-worker.ts new file mode 100644 index 0000000..7991f98 --- /dev/null +++ b/src/labeler/url-enrichment-worker.ts @@ -0,0 +1,384 @@ +import { lookup } from 'node:dns/promises' + +import { shouldRunUrlEnrichmentForDid } from '../lib/actor-pds-policy' +import { + URL_CHECK_DISCOVERY_INTERVAL_MS, + URL_CHECK_FAILED_TTL_MS, + URL_CHECK_HARD_FAILURE_ATTEMPTS, + URL_CHECK_INTERVAL_MS, + URL_CHECK_MAX_RETRY_MS, + URL_CHECK_MAX_URLS_PER_DID, + URL_CHECK_OK_TTL_MS, + URL_CHECK_RETRY_BASE_MS, + URL_CHECK_TIMEOUT_MS, + URL_ENRICHMENT_ENABLED, +} from '../lib/config' +import { + enqueueRecomputeJob, + getAllOrganizationSnapshots, + getDueUrlCheck, + getOrganizationSnapshot, + getProfileSnapshot, + getUrlCheckCounts, + getUrlResolutionMap, + recordUrlCheckFailure, + recordUrlCheckOk, + upsertPendingUrlCheck, + type UrlCheck, +} from '../lib/db' +import { isPublicNetworkAddress, normalizePublicWebsiteUrl } from '../lib/website-utils' +import type { UrlResolutionMap, UrlResolutionState } from '../lib/scoring-input' +import logger from './logger' + +const URL_RECOMPUTE_DELAY_MS = 0 +const URL_CHECK_USER_AGENT = 'orglabeler-url-enrichment/1.0' +const URL_CHECK_MAX_REDIRECTS = 5 + +type UrlCheckOutcome = + | { kind: 'ok'; statusCode: number | null } + | { kind: 'hard-failure'; statusCode: number | null; error: string } + | { kind: 'temporary-failure'; statusCode: number | null; error: string } + +class PermanentUrlFailure extends Error { + constructor(message: string) { + super(message) + this.name = 'PermanentUrlFailure' + } +} + +function retryDelayForAttempt(attempts: number): number { + const exponent = Math.max(0, attempts - 1) + return Math.min(URL_CHECK_MAX_RETRY_MS, URL_CHECK_RETRY_BASE_MS * (2 ** exponent)) +} + +function resolutionStateForCheck(check: UrlCheck): UrlResolutionState { + if (check.status === 'ok') return 'ok' + if (check.status === 'failed') return 'failed' + return 'unknown' +} + +function uniqueNormalizedUrls(urls: Array): string[] { + const normalized = urls + .map(url => normalizePublicWebsiteUrl(url)) + .filter((url): url is string => Boolean(url)) + + return [...new Set(normalized)] +} + +function collectRawUrlsForDid(did: string): Array { + const organization = getOrganizationSnapshot(did) + if (!organization) return [] + + const profile = getProfileSnapshot(did) + const urls: Array = [profile?.payload.website] + + for (const item of organization.payload.urls ?? []) { + if (urls.length >= URL_CHECK_MAX_URLS_PER_DID) break + urls.push(item?.url) + } + + return urls.slice(0, URL_CHECK_MAX_URLS_PER_DID) +} + +/** Returns the normalized public URLs currently referenced by a DID's local snapshots. */ +export function collectNormalizedUrlsForDid(did: string): string[] { + return uniqueNormalizedUrls(collectRawUrlsForDid(did)) +} + +/** + * Creates pending URL cache rows for the DID's current snapshots. + * This is safe to call after every recompute because fresh cache rows are not reset. + */ +export function enqueueUrlChecksForDid(did: string): number { + if (!URL_ENRICHMENT_ENABLED) return 0 + if (!shouldRunUrlEnrichmentForDid(did)) return 0 + + const urls = collectNormalizedUrlsForDid(did) + for (const url of urls) { + upsertPendingUrlCheck(url) + } + return urls.length +} + +/** Reads cached URL resolution states for scoring; missing rows remain optimistic unknowns. */ +export function getUrlResolutionMapForDid(did: string): UrlResolutionMap { + if (!URL_ENRICHMENT_ENABLED) return {} + + return getUrlResolutionMap(collectNormalizedUrlsForDid(did)) +} + +function discoverUrlChecksFromSnapshots(): number { + const snapshots = getAllOrganizationSnapshots() + let discovered = 0 + + for (const snapshot of snapshots) { + discovered += enqueueUrlChecksForDid(snapshot.did) + } + + return discovered +} + +function didReferencesNormalizedUrl(did: string, normalizedUrl: string): boolean { + return collectNormalizedUrlsForDid(did).includes(normalizedUrl) +} + +function getDidsReferencingUrl(normalizedUrl: string): string[] { + return getAllOrganizationSnapshots() + .map(snapshot => snapshot.did) + .filter(did => didReferencesNormalizedUrl(did, normalizedUrl)) +} + +function isOkStatus(status: number): boolean { + return (status >= 200 && status < 400) || status === 401 || status === 403 || status === 405 +} + +function isHardFailureStatus(status: number): boolean { + return status === 404 || status === 410 +} + +async function lookupHostAddresses(hostname: string): Promise> { + let timeout: NodeJS.Timeout | null = null + + try { + return await Promise.race([ + lookup(hostname, { all: true }), + new Promise((_resolve, reject) => { + timeout = setTimeout(() => reject(new Error(`DNS lookup timed out after ${URL_CHECK_TIMEOUT_MS}ms for ${hostname}`)), URL_CHECK_TIMEOUT_MS) + }), + ]) + } finally { + if (timeout) clearTimeout(timeout) + } +} + +async function assertPublicResolvedHost(normalizedUrl: string): Promise { + const hostname = new URL(normalizedUrl).hostname + + if (isPublicNetworkAddress(hostname)) return + + const addresses = await lookupHostAddresses(hostname) + if (addresses.length === 0) { + throw new PermanentUrlFailure(`DNS lookup returned no addresses for ${hostname}`) + } + + const privateAddress = addresses.find(address => !isPublicNetworkAddress(address.address)) + if (privateAddress) { + throw new PermanentUrlFailure(`URL host resolved to non-public address ${privateAddress.address}`) + } +} + +function normalizeRedirectTarget(location: string | null, currentUrl: string): string | null { + if (!location) return null + + try { + return normalizePublicWebsiteUrl(new URL(location, currentUrl).toString()) + } catch { + return null + } +} + +async function fetchUrl(url: string, method: 'HEAD' | 'GET', signal: AbortSignal): Promise { + let currentUrl = url + + for (let redirectCount = 0; redirectCount <= URL_CHECK_MAX_REDIRECTS; redirectCount++) { + await assertPublicResolvedHost(currentUrl) + + const response = await fetch(currentUrl, { + method, + redirect: 'manual', + signal, + headers: { + 'User-Agent': URL_CHECK_USER_AGENT, + Accept: 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', + ...(method === 'GET' ? { Range: 'bytes=0-0' } : {}), + }, + }) + + if (response.status < 300 || response.status >= 400) { + return response + } + + const nextUrl = normalizeRedirectTarget(response.headers.get('location'), currentUrl) + await response.body?.cancel() + + if (!nextUrl) { + throw new PermanentUrlFailure(`Unsafe or missing redirect target from ${currentUrl}`) + } + + currentUrl = nextUrl + } + + throw new PermanentUrlFailure(`Too many redirects from ${url}`) +} + +function isPermanentDnsFailure(err: unknown): boolean { + if (!err || typeof err !== 'object') return false + + const code = 'code' in err ? err.code : undefined + return code === 'ENOTFOUND' || code === 'ENODATA' || code === 'EINVAL' +} + +async function resolveUrl(normalizedUrl: string): Promise { + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), URL_CHECK_TIMEOUT_MS) + + try { + let response = await fetchUrl(normalizedUrl, 'HEAD', controller.signal) + await response.body?.cancel() + + if (response.status === 405 || response.status === 501) { + response = await fetchUrl(normalizedUrl, 'GET', controller.signal) + await response.body?.cancel() + } + + if (isOkStatus(response.status)) { + return { kind: 'ok', statusCode: response.status } + } + + if (isHardFailureStatus(response.status)) { + return { + kind: 'hard-failure', + statusCode: response.status, + error: `HTTP ${response.status}`, + } + } + + return { + kind: 'temporary-failure', + statusCode: response.status, + error: `HTTP ${response.status}`, + } + } catch (err) { + const aborted = controller.signal.aborted + const message = aborted + ? `Timed out after ${URL_CHECK_TIMEOUT_MS}ms` + : err instanceof Error + ? err.message + : String(err) + + return { + kind: err instanceof PermanentUrlFailure || isPermanentDnsFailure(err) ? 'hard-failure' : 'temporary-failure', + statusCode: null, + error: message, + } + } finally { + clearTimeout(timeout) + } +} + +function saveUrlCheckOutcome(check: UrlCheck, outcome: UrlCheckOutcome): UrlResolutionState { + if (outcome.kind === 'ok') { + recordUrlCheckOk(check.normalizedUrl, outcome.statusCode, URL_CHECK_OK_TTL_MS) + return 'ok' + } + + const attempts = check.attempts + 1 + const hardFailed = outcome.kind === 'hard-failure' && attempts >= URL_CHECK_HARD_FAILURE_ATTEMPTS + + if (hardFailed) { + recordUrlCheckFailure({ + normalizedUrl: check.normalizedUrl, + status: 'failed', + resolvable: false, + statusCode: outcome.statusCode, + error: outcome.error, + attempts, + retryAfterMs: URL_CHECK_FAILED_TTL_MS, + }) + return 'failed' + } + + recordUrlCheckFailure({ + normalizedUrl: check.normalizedUrl, + status: 'pending', + resolvable: null, + statusCode: outcome.statusCode, + error: outcome.error, + attempts, + retryAfterMs: retryDelayForAttempt(attempts), + }) + return 'unknown' +} + +async function processDueUrlCheck(check: UrlCheck): Promise { + const beforeState = resolutionStateForCheck(check) + const outcome = await resolveUrl(check.normalizedUrl) + const afterState = saveUrlCheckOutcome(check, outcome) + + logger.info( + { + url: check.normalizedUrl, + outcome: outcome.kind, + statusCode: outcome.statusCode, + attempts: outcome.kind === 'ok' ? 0 : check.attempts + 1, + beforeState, + afterState, + }, + 'Processed URL enrichment check', + ) + + if (beforeState === afterState) return + + const dids = getDidsReferencingUrl(check.normalizedUrl) + for (const did of dids) { + enqueueRecomputeJob('recompute-org', did, { + delayMs: URL_RECOMPUTE_DELAY_MS, + payload: { reason: 'url-enrichment', url: check.normalizedUrl }, + }) + } + + if (dids.length > 0) { + logger.info({ url: check.normalizedUrl, dids, beforeState, afterState }, 'Queued recompute after URL enrichment state changed') + } +} + +/** Starts the detachable URL enrichment loop that checks cached URLs outside Tap handling. */ +export function startUrlEnrichmentWorker(): { destroy: () => void } { + if (!URL_ENRICHMENT_ENABLED) { + logger.info('URL enrichment worker disabled') + return { destroy: () => undefined } + } + + let stopped = false + let running = false + let timer: NodeJS.Timeout | null = null + let lastDiscoveryAt = 0 + + const tick = async (): Promise => { + if (stopped || running) return + running = true + + try { + const now = Date.now() + if (now - lastDiscoveryAt >= URL_CHECK_DISCOVERY_INTERVAL_MS) { + const discovered = discoverUrlChecksFromSnapshots() + lastDiscoveryAt = now + if (discovered > 0) { + logger.debug({ discovered }, 'Discovered URLs from organization snapshots') + } + } + + const check = getDueUrlCheck() + if (!check) return + + await processDueUrlCheck(check) + } finally { + running = false + } + } + + timer = setInterval(() => { + tick().catch(err => logger.error({ err }, 'URL enrichment worker tick failed')) + }, URL_CHECK_INTERVAL_MS) + + logger.info({ counts: getUrlCheckCounts() }, 'URL enrichment worker started') + tick().catch(err => logger.error({ err }, 'Initial URL enrichment worker tick failed')) + + return { + destroy: () => { + stopped = true + if (timer) clearInterval(timer) + logger.info({ counts: getUrlCheckCounts() }, 'URL enrichment worker stopped') + }, + } +} diff --git a/src/lib/actor-pds-policy.ts b/src/lib/actor-pds-policy.ts new file mode 100644 index 0000000..07dd6bf --- /dev/null +++ b/src/lib/actor-pds-policy.ts @@ -0,0 +1,64 @@ +import { TEST_PDS_HOSTS } from './config' +import { + enqueueRecomputeJob, + getActorPdsCache, + isActorPdsCacheStale, + recordActorPdsPending, +} from './db' +import { isConfiguredTestPdsHost } from './pds-utils' + +/** How long a pending DID → PDS lookup should suppress duplicate refreshes. */ +export const ACTOR_PDS_PENDING_TTL_MS = 5 * 60 * 1000 + +/** How long a successful DID → PDS lookup stays fresh before refresh. */ +export const ACTOR_PDS_CACHE_TTL_MS = 24 * 60 * 60 * 1000 + +/** Returns true when actor-PDS test labeling is configured for this process. */ +export function testPdsDetectionEnabled(): boolean { + return TEST_PDS_HOSTS.length > 0 +} + +/** Enqueues actor PDS resolution when TEST_PDS_HOSTS is configured. */ +export function enqueueActorPdsResolution(did: string, reason: string, delayMs = 0): void { + if (!testPdsDetectionEnabled()) return + + recordActorPdsPending(did, ACTOR_PDS_PENDING_TTL_MS) + enqueueRecomputeJob('resolve-actor-pds', did, { + delayMs, + payload: { reason }, + }) +} + +/** + * Returns the best cached actor PDS host for scoring. Stale hosts are still used + * for the current score, but a refresh is queued so later recomputes can correct labels. + */ +export function cachedActorPdsHostForScoring(did: string): string | null { + if (!testPdsDetectionEnabled()) return null + + const cache = getActorPdsCache(did) + if (!cache || isActorPdsCacheStale(cache)) { + enqueueActorPdsResolution(did, cache ? 'recompute-stale-pds-cache' : 'recompute-pds-cache-miss') + } + + return cache?.pdsHost ?? null +} + +/** + * Decides whether URL enrichment should run for an actor. When TEST_PDS_HOSTS is + * configured, URL checks are deferred until the actor PDS cache is fresh because + * test-PDS actors are always likely-test and URL scoring is irrelevant for them. + */ +export function shouldRunUrlEnrichmentForDid(did: string): boolean { + if (!testPdsDetectionEnabled()) return true + + const cache = getActorPdsCache(did) + if (!cache || isActorPdsCacheStale(cache)) { + enqueueActorPdsResolution(did, cache ? 'url-enrichment-stale-pds-cache' : 'url-enrichment-pds-cache-miss') + return false + } + + if (cache.status !== 'ok' || !cache.pdsHost) return false + + return !isConfiguredTestPdsHost(cache.pdsHost, TEST_PDS_HOSTS) +} diff --git a/src/lib/config.ts b/src/lib/config.ts index 646b92f..32c25a0 100644 --- a/src/lib/config.ts +++ b/src/lib/config.ts @@ -1,3 +1,5 @@ +import { parseTestPdsHosts } from './pds-utils' + export const DID = process.env.DID ?? '' export const SIGNING_KEY = process.env.SIGNING_KEY ?? '' export const HOST = process.env.HOST ?? '0.0.0.0' @@ -18,6 +20,37 @@ export const TAP_ADMIN_PASSWORD = process.env.TAP_ADMIN_PASSWORD ?? '' export const HF_TOKEN = process.env.HF_TOKEN ?? '' export const HF_MODEL = 'facebook/bart-large-mnli' export const HYPERSCAN_RECORD_URL_BASE = process.env.HYPERSCAN_RECORD_URL_BASE ?? 'https://hyperscan.dev/data' +/** Comma-separated PDS hosts whose actors should always be labeled likely-test. */ +export const TEST_PDS_HOSTS = parseTestPdsHosts(process.env.TEST_PDS_HOSTS ?? '') + +function integerEnv(name: string, fallback: number): number { + const value = process.env[name] + if (!value) return fallback + + const parsed = Number(value) + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback +} + +/** Enables the detachable URL enrichment worker; set to false to keep scoring fully provisional. */ +export const URL_ENRICHMENT_ENABLED = process.env.URL_ENRICHMENT_ENABLED !== 'false' +/** Poll interval for checking one due URL cache row. */ +export const URL_CHECK_INTERVAL_MS = integerEnv('URL_CHECK_INTERVAL_MS', 1000) +/** How often the URL worker scans local snapshots for newly referenced URLs. */ +export const URL_CHECK_DISCOVERY_INTERVAL_MS = integerEnv('URL_CHECK_DISCOVERY_INTERVAL_MS', 30_000) +/** Maximum wall-clock time for a single URL resolution attempt. */ +export const URL_CHECK_TIMEOUT_MS = integerEnv('URL_CHECK_TIMEOUT_MS', 4000) +/** How long a successful URL resolution remains fresh before rechecking. */ +export const URL_CHECK_OK_TTL_MS = integerEnv('URL_CHECK_OK_TTL_MS', 7 * 24 * 60 * 60 * 1000) +/** How long a hard failed URL remains downgraded before another attempt. */ +export const URL_CHECK_FAILED_TTL_MS = integerEnv('URL_CHECK_FAILED_TTL_MS', 24 * 60 * 60 * 1000) +/** Initial retry delay for temporary URL failures. */ +export const URL_CHECK_RETRY_BASE_MS = integerEnv('URL_CHECK_RETRY_BASE_MS', 5 * 60 * 1000) +/** Maximum retry delay for temporary URL failures. */ +export const URL_CHECK_MAX_RETRY_MS = integerEnv('URL_CHECK_MAX_RETRY_MS', 60 * 60 * 1000) +/** Number of hard failures required before URL scoring removes resolve points. */ +export const URL_CHECK_HARD_FAILURE_ATTEMPTS = integerEnv('URL_CHECK_HARD_FAILURE_ATTEMPTS', 2) +/** Maximum profile/organization URLs to cache and check per DID. */ +export const URL_CHECK_MAX_URLS_PER_DID = integerEnv('URL_CHECK_MAX_URLS_PER_DID', 5) export function validateLabelerConfig(): void { const required: [string, string][] = [ diff --git a/src/lib/constants.ts b/src/lib/constants.ts index 1be9b1d..e348ad0 100644 --- a/src/lib/constants.ts +++ b/src/lib/constants.ts @@ -62,8 +62,8 @@ export const TEST_PATTERNS: RegExp[] = [ // Word-boundary "test" — catches "Another Test", "Test Contributors", "This is testing", "test 123" /\btest(ing|ed|er|s)?\b/i, - // Common junk prefixes - /^asdf/i, /^lorem ipsum/i, /^placeholder/i, /^delete me/i, /^ignore/i, /^zzz/i, + // Common junk prefixes and phrases. + /^asdf/i, /\blorem ipsum\b/i, /^placeholder/i, /^delete me/i, /^ignore/i, /^zzz/i, // Exact match common junk /^foo$/i, /^bar$/i, /^abc$/i, /^123$/i, /^wip$/i, /^todo$/i, @@ -95,5 +95,28 @@ export const AUTHENTICITY_TEXT_PATTERNS: RegExp[] = [ /^unlisted$/i, ] +/** + * Extra authenticity patterns for display names, where workflow/test labels are + * much stronger evidence than the same words appearing in longer descriptions. + */ +export const DISPLAY_NAME_AUTHENTICITY_TEXT_PATTERNS: RegExp[] = [ + ...AUTHENTICITY_TEXT_PATTERNS, + /(?:^|[^\p{Letter}\p{Number}])(?:demo|dev|staging|qa|e2e|sandbox|fixture)(?:$|[^\p{Letter}\p{Number}])/iu, + /^tobytest\d*$/i, + /^exclusivecgstester\d*$/i, + /\b(?:seed data|new db|changes requested)\b/i, + /^unpublished(?:\s+org(?:anization)?)?$/i, + /^published(?:\s+org(?:anization)?)?$/i, + /^org(?:anization)?(?:[\s._-]*\d+)?$/i, + /^(?:my\s+)?first\s+org(?:anization)?$/i, + /^new[\s._-]+org(?:anization)?$/i, +] + +/** Reserved example domains that should never count as organization evidence. */ +export const PLACEHOLDER_DOMAINS = ['example.com', 'example.net', 'example.org'] as const + +/** Reserved non-production TLDs that should fail the authenticity gate. */ +export const PLACEHOLDER_TLDS = ['example', 'invalid', 'test'] as const + export const LABEL_LIMIT = 1 export const QUALITY_LABEL_IDENTIFIERS: string[] = ['likely-test', 'standard', 'high-quality'] diff --git a/src/lib/db.ts b/src/lib/db.ts index b9f3f0a..f8bd339 100644 --- a/src/lib/db.ts +++ b/src/lib/db.ts @@ -8,6 +8,7 @@ import type { ProfileSnapshot, RuntimeLabelTier, } from './types' +import type { UrlResolutionMap, UrlResolutionState } from './scoring-input' export const HF_POSITIVE_LABEL = 'well-formed actor profile' @@ -49,11 +50,100 @@ interface PendingOrganizationDeleteRow { rkey: string attempts: number last_attempt_at: string | null + next_attempt_at: string | null last_error: string | null created_at: string updated_at: string } +export type RecomputeJobKind = 'recompute-org' | 'resolve-actor-pds' +export type RecomputeJobStatus = 'pending' | 'running' | 'done' | 'failed' + +export interface RecomputeJob { + id: number + kind: RecomputeJobKind + key: string + status: RecomputeJobStatus + attempts: number + runAfter: string + payload: string | null + lastError: string | null + createdAt: string + updatedAt: string +} + +interface RecomputeJobRow { + id: number + kind: string + key: string + status: string + attempts: number + run_after: string + payload: string | null + last_error: string | null + created_at: string + updated_at: string +} + +export type ActorPdsCacheStatus = 'pending' | 'ok' | 'failed' + +/** Cached DID → PDS resolution state used by test-PDS labeling and URL gating. */ +export interface ActorPdsCache { + did: string + status: ActorPdsCacheStatus + pdsUrl: string | null + pdsHost: string | null + checkedAt: string | null + expiresAt: string + lastError: string | null + createdAt: string + updatedAt: string +} + +interface ActorPdsCacheRow { + did: string + status: string + pds_url: string | null + pds_host: string | null + checked_at: string | null + expires_at: string + last_error: string | null + created_at: string + updated_at: string +} + +/** Durable URL cache status. Pending means scoring should keep optimistic provisional URL points. */ +export type UrlCheckStatus = 'pending' | 'ok' | 'failed' + +/** URL cache row used by the async URL enrichment worker. */ +export interface UrlCheck { + normalizedUrl: string + status: UrlCheckStatus + resolvable: boolean | null + statusCode: number | null + error: string | null + attempts: number + lastAttemptAt: string | null + checkedAt: string | null + expiresAt: string + createdAt: string + updatedAt: string +} + +interface UrlCheckRow { + normalized_url: string + status: string + resolvable: number | null + status_code: number | null + error: string | null + attempts: number + last_attempt_at: string | null + checked_at: string | null + expires_at: string + created_at: string + updated_at: string +} + type ActivityLogInput = { did: string rkey: string @@ -99,6 +189,67 @@ function createActivitiesTable(db: Database.Database): void { `) } +function createRecomputeJobsTable(db: Database.Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS recompute_jobs ( + id INTEGER PRIMARY KEY, + kind TEXT NOT NULL, + key TEXT NOT NULL, + status TEXT NOT NULL CHECK(status IN ('pending', 'running', 'done', 'failed')), + attempts INTEGER NOT NULL DEFAULT 0, + run_after TEXT NOT NULL, + payload TEXT, + last_error TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')), + UNIQUE(kind, key) + ); + + CREATE INDEX IF NOT EXISTS idx_recompute_jobs_due ON recompute_jobs(status, run_after); + CREATE INDEX IF NOT EXISTS idx_recompute_jobs_updated_at ON recompute_jobs(updated_at); + `) +} + +function createActorPdsCacheTable(db: Database.Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS actor_pds_cache ( + did TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK(status IN ('pending', 'ok', 'failed')), + pds_url TEXT, + pds_host TEXT, + checked_at TEXT, + expires_at TEXT NOT NULL, + last_error TEXT, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + + CREATE INDEX IF NOT EXISTS idx_actor_pds_cache_status_expires ON actor_pds_cache(status, expires_at); + CREATE INDEX IF NOT EXISTS idx_actor_pds_cache_host ON actor_pds_cache(pds_host); + `) +} + +function createUrlChecksTable(db: Database.Database): void { + db.exec(` + CREATE TABLE IF NOT EXISTS url_checks ( + normalized_url TEXT PRIMARY KEY, + status TEXT NOT NULL CHECK(status IN ('pending', 'ok', 'failed')), + resolvable INTEGER, + status_code INTEGER, + error TEXT, + attempts INTEGER NOT NULL DEFAULT 0, + last_attempt_at TEXT, + checked_at TEXT, + expires_at TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + + CREATE INDEX IF NOT EXISTS idx_url_checks_due ON url_checks(status, expires_at); + CREATE INDEX IF NOT EXISTS idx_url_checks_updated_at ON url_checks(updated_at); + `) +} + function createSnapshotTables(db: Database.Database): void { db.exec(` CREATE TABLE IF NOT EXISTS profile_snapshots ( @@ -128,6 +279,7 @@ function createSnapshotTables(db: Database.Database): void { rkey TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, last_attempt_at TEXT, + next_attempt_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now')), updated_at TEXT NOT NULL DEFAULT (datetime('now')) @@ -146,6 +298,9 @@ export function getDb(): Database.Database { createActivitiesTable(_db) createSnapshotTables(_db) + createRecomputeJobsTable(_db) + createActorPdsCacheTable(_db) + createUrlChecksTable(_db) // Migration: recreate tables whose tier CHECK constraint predates the active 3-tier set. try { @@ -193,6 +348,12 @@ export function getDb(): Database.Database { _db.exec("ALTER TABLE profile_snapshots ADD COLUMN validation_notes TEXT NOT NULL DEFAULT '[]'") } + const pendingDeleteCols = (_db.prepare("PRAGMA table_info(pending_organization_deletes)").all() as Array<{ name: string }>).map(c => c.name) + if (!pendingDeleteCols.includes('next_attempt_at')) { + _db.exec('ALTER TABLE pending_organization_deletes ADD COLUMN next_attempt_at TEXT') + } + _db.exec('CREATE INDEX IF NOT EXISTS idx_pending_organization_deletes_next_attempt ON pending_organization_deletes(next_attempt_at)') + return _db } @@ -351,39 +512,81 @@ export function deleteOrganizationRecordState(did: string, rkey: string): void { db.prepare('DELETE FROM activities WHERE did = ? AND rkey = ?').run(did, rkey) } +/** + * Completes a queued organization delete without deleting a newer replacement snapshot. + * Returns false when the pending delete was already cleared or superseded by a newer upsert. + */ +export function completePendingOrganizationDelete(did: string, rkey: string, recordUri: string): boolean { + const db = getDb() + + return db.transaction(() => { + const pending = db.prepare(` + SELECT did, record_uri, rkey + FROM pending_organization_deletes + WHERE did = ? + `).get(did) as Pick | undefined + + if (!pending || pending.rkey !== rkey || pending.record_uri !== recordUri) { + return false + } + + db.prepare(` + DELETE FROM organization_snapshots + WHERE did = @did AND rkey = @rkey AND record_uri = @recordUri + `).run({ did, rkey, recordUri }) + db.prepare('DELETE FROM activities WHERE did = ? AND rkey = ?').run(did, rkey) + db.prepare(` + DELETE FROM pending_organization_deletes + WHERE did = @did AND rkey = @rkey AND record_uri = @recordUri + `).run({ did, rkey, recordUri }) + + return true + })() +} + export function upsertPendingOrganizationDelete(did: string, rkey: string, recordUri: string): void { const db = getDb() db.prepare(` INSERT INTO pending_organization_deletes - (did, record_uri, rkey, attempts, last_attempt_at, last_error, created_at, updated_at) + (did, record_uri, rkey, attempts, last_attempt_at, next_attempt_at, last_error, created_at, updated_at) VALUES - (@did, @recordUri, @rkey, 0, NULL, NULL, datetime('now'), datetime('now')) + (@did, @recordUri, @rkey, 0, NULL, NULL, NULL, datetime('now'), datetime('now')) ON CONFLICT(did) DO UPDATE SET record_uri = excluded.record_uri, rkey = excluded.rkey, + next_attempt_at = NULL, updated_at = excluded.updated_at `).run({ did, rkey, recordUri }) } -export function recordPendingOrganizationDeleteAttempt(did: string, errorMessage?: string | null): void { +export function recordPendingOrganizationDeleteAttempt(did: string, errorMessage?: string | null, retryDelayMs = 60_000): void { const db = getDb() + const nextAttemptAt = new Date(Date.now() + retryDelayMs).toISOString() db.prepare(` UPDATE pending_organization_deletes SET attempts = attempts + 1, last_attempt_at = datetime('now'), + next_attempt_at = @nextAttemptAt, last_error = @lastError, updated_at = datetime('now') WHERE did = @did - `).run({ did, lastError: errorMessage ?? null }) + `).run({ did, nextAttemptAt, lastError: errorMessage ?? null }) } -export function getPendingOrganizationDeletes(): Array { +export function getPendingOrganizationDeletes(now = new Date().toISOString()): Array { const db = getDb() return db.prepare(` - SELECT did, record_uri, rkey, attempts, last_attempt_at, last_error, created_at, updated_at + SELECT did, record_uri, rkey, attempts, last_attempt_at, next_attempt_at, last_error, created_at, updated_at FROM pending_organization_deletes + WHERE next_attempt_at IS NULL OR next_attempt_at <= @now ORDER BY updated_at ASC - `).all() as Array + `).all({ now }) as Array +} + +export function hasPendingOrganizationDelete(did: string): boolean { + const db = getDb() + const row = db.prepare('SELECT 1 FROM pending_organization_deletes WHERE did = ? LIMIT 1').get(did) + return Boolean(row) } export function deletePendingOrganizationDelete(did: string): void { @@ -391,6 +594,361 @@ export function deletePendingOrganizationDelete(did: string): void { db.prepare('DELETE FROM pending_organization_deletes WHERE did = ?').run(did) } +/** + * Coalesces actor-level scoring work into one durable SQLite job per DID. + * Use this from the Tap handler after persisting snapshots so slow label work can + * run after Tap has acknowledged the event. + */ +export function enqueueRecomputeJob( + kind: RecomputeJobKind, + key: string, + options: { delayMs?: number; payload?: Record | null } = {}, +): void { + const db = getDb() + const nowMs = Date.now() + const runAfter = new Date(nowMs + (options.delayMs ?? 2000)).toISOString() + const payload = options.payload === undefined || options.payload === null + ? null + : JSON.stringify(options.payload) + + db.prepare(` + INSERT INTO recompute_jobs + (kind, key, status, attempts, run_after, payload, last_error, created_at, updated_at) + VALUES + (@kind, @key, 'pending', 0, @runAfter, @payload, NULL, datetime('now'), datetime('now')) + ON CONFLICT(kind, key) DO UPDATE SET + status = 'pending', + attempts = CASE WHEN recompute_jobs.status IN ('done', 'failed') THEN 0 ELSE recompute_jobs.attempts END, + run_after = excluded.run_after, + payload = COALESCE(excluded.payload, recompute_jobs.payload), + last_error = NULL, + updated_at = datetime('now') + `).run({ kind, key, runAfter, payload }) +} + +/** + * Atomically claims the oldest due recompute job for a single worker process. + * Returns null when no debounced job is ready to run yet. + */ +export function claimDueRecomputeJob(now = new Date().toISOString()): RecomputeJob | null { + const db = getDb() + const row = db.prepare(` + UPDATE recompute_jobs + SET status = 'running', + attempts = attempts + 1, + updated_at = datetime('now') + WHERE id = ( + SELECT id + FROM recompute_jobs + WHERE status = 'pending' AND run_after <= @now + ORDER BY run_after ASC, updated_at ASC + LIMIT 1 + ) + RETURNING id, kind, key, status, attempts, run_after, payload, last_error, created_at, updated_at + `).get({ now }) as RecomputeJobRow | undefined + + return row ? recomputeJobRowToEntry(row) : null +} + +/** + * Marks a claimed recompute job as done while keeping the row available for metrics. + * The status guard preserves a newer pending enqueue that arrived while this job was running. + */ +export function completeRecomputeJob(id: number): number { + const db = getDb() + const result = db.prepare(` + UPDATE recompute_jobs + SET status = 'done', + attempts = 0, + last_error = NULL, + updated_at = datetime('now') + WHERE id = ? AND status = 'running' + `).run(id) + + return result.changes +} + +/** + * Records a recompute failure and schedules another attempt with bounded backoff. + * The unique job row is reused, so newer Tap events can still coalesce over it. + */ +export function recoverStaleRunningRecomputeJobs(staleAfterMs: number): number { + const db = getDb() + const staleSeconds = Math.max(1, Math.ceil(staleAfterMs / 1000)) + const result = db.prepare(` + UPDATE recompute_jobs + SET status = 'pending', + run_after = @now, + last_error = 'Recovered stale running recompute job after process interruption', + updated_at = datetime('now') + WHERE status = 'running' AND updated_at <= datetime('now', @staleModifier) + `).run({ now: new Date().toISOString(), staleModifier: `-${staleSeconds} seconds` }) + + return result.changes +} + +export function failRecomputeJob( + id: number, + errorMessage: string, + retryDelayMs: number, + status: Extract = 'pending', +): void { + const db = getDb() + const runAfter = new Date(Date.now() + retryDelayMs).toISOString() + db.prepare(` + UPDATE recompute_jobs + SET status = @status, + run_after = @runAfter, + last_error = @errorMessage, + updated_at = datetime('now') + WHERE id = @id AND status = 'running' + `).run({ id, status, runAfter, errorMessage }) +} + +/** Returns queue counts grouped by status for lightweight health logging and metrics. */ +export function getRecomputeJobCounts(): Record { + const db = getDb() + const counts: Record = { + pending: 0, + running: 0, + done: 0, + failed: 0, + } + const rows = db.prepare(` + SELECT status, COUNT(*) as count + FROM recompute_jobs + GROUP BY status + `).all() as Array<{ status: RecomputeJobStatus; count: number }> + + for (const row of rows) { + counts[row.status] = row.count + } + return counts +} + +/** Reads cached actor PDS state, if this DID has been resolved before. */ +export function getActorPdsCache(did: string): ActorPdsCache | null { + const db = getDb() + const row = db.prepare(` + SELECT did, status, pds_url, pds_host, checked_at, expires_at, last_error, created_at, updated_at + FROM actor_pds_cache + WHERE did = ? + `).get(did) as ActorPdsCacheRow | undefined + + return row ? actorPdsCacheRowToEntry(row) : null +} + +/** Returns true when a cached PDS row should be refreshed before URL enrichment. */ +export function isActorPdsCacheStale(cache: ActorPdsCache, now = new Date().toISOString()): boolean { + return cache.expiresAt <= now +} + +/** Creates or refreshes a pending actor PDS cache row without erasing old host data. */ +export function recordActorPdsPending(did: string, retryAfterMs: number): void { + const db = getDb() + const expiresAt = new Date(Date.now() + retryAfterMs).toISOString() + db.prepare(` + INSERT INTO actor_pds_cache + (did, status, pds_url, pds_host, checked_at, expires_at, last_error, created_at, updated_at) + VALUES + (@did, 'pending', NULL, NULL, NULL, @expiresAt, NULL, datetime('now'), datetime('now')) + ON CONFLICT(did) DO UPDATE SET + status = 'pending', + expires_at = excluded.expires_at, + last_error = NULL, + updated_at = datetime('now') + `).run({ did, expiresAt }) +} + +/** Stores a successful actor DID → PDS resolution result. */ +export function recordActorPdsOk(did: string, pdsUrl: string, pdsHost: string, ttlMs: number): void { + const db = getDb() + const now = new Date().toISOString() + const expiresAt = new Date(Date.now() + ttlMs).toISOString() + db.prepare(` + INSERT INTO actor_pds_cache + (did, status, pds_url, pds_host, checked_at, expires_at, last_error, created_at, updated_at) + VALUES + (@did, 'ok', @pdsUrl, @pdsHost, @now, @expiresAt, NULL, datetime('now'), datetime('now')) + ON CONFLICT(did) DO UPDATE SET + status = 'ok', + pds_url = excluded.pds_url, + pds_host = excluded.pds_host, + checked_at = excluded.checked_at, + expires_at = excluded.expires_at, + last_error = NULL, + updated_at = datetime('now') + `).run({ did, pdsUrl, pdsHost, now, expiresAt }) +} + +/** Stores a failed actor PDS resolution attempt while preserving any stale host. */ +export function recordActorPdsFailure(did: string, errorMessage: string, retryAfterMs: number): void { + const db = getDb() + const now = new Date().toISOString() + const expiresAt = new Date(Date.now() + retryAfterMs).toISOString() + db.prepare(` + INSERT INTO actor_pds_cache + (did, status, pds_url, pds_host, checked_at, expires_at, last_error, created_at, updated_at) + VALUES + (@did, 'failed', NULL, NULL, @now, @expiresAt, @errorMessage, datetime('now'), datetime('now')) + ON CONFLICT(did) DO UPDATE SET + status = 'failed', + checked_at = excluded.checked_at, + expires_at = excluded.expires_at, + last_error = excluded.last_error, + updated_at = datetime('now') + `).run({ did, now, expiresAt, errorMessage }) +} + +/** + * Ensures a normalized public URL has a cache row for async enrichment. + * Fresh ok/failed rows are left untouched so repeated snapshots do not reset TTLs. + */ +export function upsertPendingUrlCheck(normalizedUrl: string, now = new Date().toISOString()): void { + const db = getDb() + db.prepare(` + INSERT INTO url_checks + (normalized_url, status, resolvable, status_code, error, attempts, last_attempt_at, checked_at, expires_at, created_at, updated_at) + VALUES + (@normalizedUrl, 'pending', NULL, NULL, NULL, 0, NULL, NULL, @now, datetime('now'), datetime('now')) + ON CONFLICT(normalized_url) DO UPDATE SET + status = CASE WHEN url_checks.expires_at <= @now AND url_checks.status != 'pending' THEN 'pending' ELSE url_checks.status END, + resolvable = CASE WHEN url_checks.expires_at <= @now AND url_checks.status != 'pending' THEN NULL ELSE url_checks.resolvable END, + status_code = CASE WHEN url_checks.expires_at <= @now AND url_checks.status != 'pending' THEN NULL ELSE url_checks.status_code END, + error = CASE WHEN url_checks.expires_at <= @now AND url_checks.status != 'pending' THEN NULL ELSE url_checks.error END, + attempts = CASE WHEN url_checks.expires_at <= @now AND url_checks.status != 'pending' THEN 0 ELSE url_checks.attempts END, + expires_at = CASE WHEN url_checks.expires_at <= @now AND url_checks.status != 'pending' THEN @now ELSE url_checks.expires_at END, + updated_at = CASE WHEN url_checks.expires_at <= @now AND url_checks.status != 'pending' THEN datetime('now') ELSE url_checks.updated_at END + `).run({ normalizedUrl, now }) +} + +/** Returns the next due URL cache row, or null when no URL check is ready. */ +export function getDueUrlCheck(now = new Date().toISOString()): UrlCheck | null { + const db = getDb() + const row = db.prepare(` + SELECT normalized_url, status, resolvable, status_code, error, attempts, last_attempt_at, checked_at, expires_at, created_at, updated_at + FROM url_checks + WHERE expires_at <= @now + ORDER BY CASE status WHEN 'pending' THEN 0 ELSE 1 END, expires_at ASC, updated_at ASC + LIMIT 1 + `).get({ now }) as UrlCheckRow | undefined + + return row ? urlCheckRowToEntry(row) : null +} + +/** Stores a successful URL check and keeps the ok result fresh for the supplied TTL. */ +export function recordUrlCheckOk(normalizedUrl: string, statusCode: number | null, ttlMs: number): void { + const db = getDb() + const now = new Date() + const checkedAt = now.toISOString() + const expiresAt = new Date(now.getTime() + ttlMs).toISOString() + + db.prepare(` + UPDATE url_checks + SET status = 'ok', + resolvable = 1, + status_code = @statusCode, + error = NULL, + attempts = 0, + last_attempt_at = @checkedAt, + checked_at = @checkedAt, + expires_at = @expiresAt, + updated_at = datetime('now') + WHERE normalized_url = @normalizedUrl + `).run({ normalizedUrl, statusCode, checkedAt, expiresAt }) +} + +/** + * Stores a failed URL check. Use status='pending' for temporary retryable + * failures and status='failed' only when scoring should remove URL resolve points. + */ +export function recordUrlCheckFailure(options: { + normalizedUrl: string + status: Extract + resolvable: boolean | null + statusCode: number | null + error: string + attempts: number + retryAfterMs: number +}): void { + const db = getDb() + const now = new Date() + const checkedAt = now.toISOString() + const expiresAt = new Date(now.getTime() + options.retryAfterMs).toISOString() + + db.prepare(` + UPDATE url_checks + SET status = @status, + resolvable = @resolvable, + status_code = @statusCode, + error = @error, + attempts = @attempts, + last_attempt_at = @checkedAt, + checked_at = @checkedAt, + expires_at = @expiresAt, + updated_at = datetime('now') + WHERE normalized_url = @normalizedUrl + `).run({ + normalizedUrl: options.normalizedUrl, + status: options.status, + resolvable: options.resolvable === null ? null : options.resolvable ? 1 : 0, + statusCode: options.statusCode, + error: options.error.slice(0, 1000), + attempts: options.attempts, + checkedAt, + expiresAt, + }) +} + +/** Returns the active scoring state for normalized URLs from the detachable URL cache. */ +export function getUrlResolutionMap(normalizedUrls: string[], now = new Date().toISOString()): UrlResolutionMap { + const db = getDb() + const uniqueUrls = [...new Set(normalizedUrls)].filter(url => url.length > 0) + if (uniqueUrls.length === 0) return {} + + const placeholders = uniqueUrls.map(() => '?').join(', ') + const rows = db.prepare(` + SELECT normalized_url, status, expires_at + FROM url_checks + WHERE normalized_url IN (${placeholders}) AND expires_at > ? + `).all(...uniqueUrls, now) as Array<{ normalized_url: string; status: UrlCheckStatus; expires_at: string }> + + const states: UrlResolutionMap = {} + for (const row of rows) { + const state = urlCheckStatusToResolutionState(row.status) + if (state !== 'unknown') { + states[row.normalized_url] = state + } + } + return states +} + +/** Returns URL cache counts grouped by status for metrics and smoke tests. */ +export function getUrlCheckCounts(): Record { + const db = getDb() + const counts: Record = { + pending: 0, + ok: 0, + failed: 0, + } + const rows = db.prepare(` + SELECT status, COUNT(*) as count + FROM url_checks + GROUP BY status + `).all() as Array<{ status: UrlCheckStatus; count: number }> + + for (const row of rows) { + counts[row.status] = row.count + } + return counts +} + +function urlCheckStatusToResolutionState(status: UrlCheckStatus): UrlResolutionState { + if (status === 'ok') return 'ok' + if (status === 'failed') return 'failed' + return 'unknown' +} + // Upsert on did+rkey. Preserves existing hf_label/hf_score when the new values are null. export function logActivity(entry: ActivityLogInput, hf?: HfClassificationData): void { const db = getDb() @@ -641,6 +1199,51 @@ export function getActivityByDidRkey(did: string, rkey: string): ActivityLogEntr return row ? rowToEntry(row) : null } +function actorPdsCacheRowToEntry(row: ActorPdsCacheRow): ActorPdsCache { + return { + did: row.did, + status: row.status as ActorPdsCacheStatus, + pdsUrl: row.pds_url, + pdsHost: row.pds_host, + checkedAt: row.checked_at, + expiresAt: row.expires_at, + lastError: row.last_error, + createdAt: row.created_at, + updatedAt: row.updated_at, + } +} + +function recomputeJobRowToEntry(row: RecomputeJobRow): RecomputeJob { + return { + id: row.id, + kind: row.kind as RecomputeJobKind, + key: row.key, + status: row.status as RecomputeJobStatus, + attempts: row.attempts, + runAfter: row.run_after, + payload: row.payload, + lastError: row.last_error, + createdAt: row.created_at, + updatedAt: row.updated_at, + } +} + +function urlCheckRowToEntry(row: UrlCheckRow): UrlCheck { + return { + normalizedUrl: row.normalized_url, + status: row.status as UrlCheckStatus, + resolvable: row.resolvable === null ? null : row.resolvable === 1, + statusCode: row.status_code, + error: row.error, + attempts: row.attempts, + lastAttemptAt: row.last_attempt_at, + checkedAt: row.checked_at, + expiresAt: row.expires_at, + createdAt: row.created_at, + updatedAt: row.updated_at, + } +} + function rowToEntry(row: Record): ActivityLogEntry { return { id: row['id'] as number, diff --git a/src/lib/pds-test-override.ts b/src/lib/pds-test-override.ts new file mode 100644 index 0000000..7d07705 --- /dev/null +++ b/src/lib/pds-test-override.ts @@ -0,0 +1,46 @@ +import { TEST_PDS_HOSTS } from './config' +import { isConfiguredTestPdsHost, normalizePdsHost } from './pds-utils' +import { tierForScore } from './scorer' +import type { ScoreResult } from './types' + +/** Stable prefix for test signals created from configured actor PDS hosts. */ +export const PDS_TEST_SIGNAL_PREFIX = 'actor-pds-test-host:' + +/** Builds the dashboard-visible test signal for a matching actor PDS host. */ +export function pdsTestSignal(pdsHost: string): string { + return `${PDS_TEST_SIGNAL_PREFIX} ${pdsHost}` +} + +/** Removes stale PDS signals and adds the current one when the host is configured as test. */ +export function updatePdsTestSignals( + testSignals: string[], + pdsHost: string | null | undefined, + testPdsHosts: readonly string[] = TEST_PDS_HOSTS, +): string[] { + const withoutPdsSignals = testSignals.filter(signal => !signal.startsWith(PDS_TEST_SIGNAL_PREFIX)) + const normalizedHost = pdsHost ? normalizePdsHost(pdsHost) : null + + if (normalizedHost === null || !isConfiguredTestPdsHost(normalizedHost, testPdsHosts)) { + return withoutPdsSignals + } + + return [...withoutPdsSignals, pdsTestSignal(normalizedHost)] +} + +/** + * Applies the configured actor-PDS test override to a score result. The numeric + * score and breakdown stay unchanged; only test signals and the derived tier can change. + */ +export function applyPdsTestOverride( + result: TScore, + pdsHost: string | null | undefined, + testPdsHosts: readonly string[] = TEST_PDS_HOSTS, +): TScore { + const updatedSignals = updatePdsTestSignals(result.testSignals, pdsHost, testPdsHosts) + + return { + ...result, + tier: tierForScore(result.totalScore, updatedSignals), + testSignals: updatedSignals, + } as TScore +} diff --git a/src/lib/pds-utils.ts b/src/lib/pds-utils.ts new file mode 100644 index 0000000..d1b15a3 --- /dev/null +++ b/src/lib/pds-utils.ts @@ -0,0 +1,41 @@ +/** + * Normalizes AT Protocol PDS hostnames for exact configuration matching. + * Inputs may be bare hosts or full PDS endpoint URLs. + */ +export function normalizePdsHost(value: string): string | null { + const trimmed = value.trim() + if (!trimmed) return null + + try { + const url = trimmed.includes('://') + ? new URL(trimmed) + : new URL(`https://${trimmed}`) + return url.hostname.toLowerCase().replace(/\.+$/, '') || null + } catch { + return null + } +} + +/** Returns the normalized hostname for a resolved PDS endpoint URL. */ +export function normalizePdsHostFromUrl(pdsUrl: string): string | null { + return normalizePdsHost(pdsUrl) +} + +/** Parses a comma-separated TEST_PDS_HOSTS value into unique normalized hosts. */ +export function parseTestPdsHosts(value: string): string[] { + const hosts = value + .split(',') + .map(host => normalizePdsHost(host)) + .filter((host): host is string => host !== null) + + return [...new Set(hosts)] +} + +/** Checks whether a PDS host exactly matches one of the configured test hosts. */ +export function isConfiguredTestPdsHost( + pdsHost: string | null | undefined, + testPdsHosts: readonly string[], +): boolean { + const normalized = pdsHost ? normalizePdsHost(pdsHost) : null + return normalized !== null && testPdsHosts.includes(normalized) +} diff --git a/src/lib/resolve-pds.ts b/src/lib/resolve-pds.ts index 7e75035..c2d9ae8 100644 --- a/src/lib/resolve-pds.ts +++ b/src/lib/resolve-pds.ts @@ -1,30 +1,67 @@ +interface DidDocument { + service?: Array<{ id: string; type?: string; serviceEndpoint?: unknown }> +} + +function extractPdsEndpoint(did: string, didDoc: DidDocument): string { + const pdsService = didDoc.service?.find( + service => service.id === '#atproto_pds' || service.id.endsWith('#atproto_pds') + ) + + if (typeof pdsService?.serviceEndpoint !== 'string') { + throw new Error(`No #atproto_pds service found in DID document for ${did}`) + } + + return pdsService.serviceEndpoint +} + +function didWebDocumentUrl(did: string): string { + const methodSpecificId = did.slice('did:web:'.length) + const parts = methodSpecificId.split(':').map(part => decodeURIComponent(part)) + const host = parts[0] + if (!host) throw new Error(`Invalid did:web identifier: ${did}`) + + if (parts.length === 1) { + return `https://${host}/.well-known/did.json` + } + + return `https://${host}/${parts.slice(1).join('/')}/did.json` +} + +async function fetchDidDocument(did: string): Promise { + let url: string + if (did.startsWith('did:plc:')) { + url = `https://plc.directory/${did}` + } else if (did.startsWith('did:web:')) { + url = didWebDocumentUrl(did) + } else { + throw new Error(`Unsupported DID method for PDS resolution: ${did}`) + } + + const res = await fetch(url) + if (!res.ok) { + throw new Error(`Failed to fetch DID document for ${did}: HTTP ${res.status}`) + } + + return (await res.json()) as DidDocument +} + +/** Resolves the actor PDS endpoint from a DID document. */ +export async function resolvePdsForDid(did: string): Promise<{ did: string; pds: string }> { + const didDoc = await fetchDidDocument(did) + return { did, pds: extractPdsEndpoint(did, didDoc) } +} + /** * Resolve the PDS endpoint for a handle. - * Steps: handle → DID → PLC directory → extract #atproto_pds serviceEndpoint + * Steps: handle → DID → DID document → extract #atproto_pds serviceEndpoint. */ export async function resolvePds(handle: string): Promise<{ did: string; pds: string }> { - // Step 1: Resolve handle to DID const handleRes = await fetch( `https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle?handle=${encodeURIComponent(handle)}` ) if (!handleRes.ok) throw new Error(`Failed to resolve handle: HTTP ${handleRes.status}`) const { did } = (await handleRes.json()) as { did: string } - if (!did) throw new Error("No DID returned for handle") - - // Step 2: Look up DID document from PLC directory - const plcRes = await fetch(`https://plc.directory/${did}`) - if (!plcRes.ok) throw new Error(`Failed to fetch DID document: HTTP ${plcRes.status}`) - const didDoc = (await plcRes.json()) as { - service?: Array<{ id: string; type: string; serviceEndpoint: string }> - } - - // Step 3: Extract PDS endpoint - const pdsService = didDoc.service?.find( - (s) => s.id === "#atproto_pds" || s.id.endsWith("#atproto_pds") - ) - if (!pdsService?.serviceEndpoint) { - throw new Error(`No #atproto_pds service found in DID document for ${did}`) - } + if (!did) throw new Error('No DID returned for handle') - return { did, pds: pdsService.serviceEndpoint } + return resolvePdsForDid(did) } diff --git a/src/lib/scorer.ts b/src/lib/scorer.ts index ee36dd1..9806922 100644 --- a/src/lib/scorer.ts +++ b/src/lib/scorer.ts @@ -1,7 +1,6 @@ import type { LabelTier, ScoreBreakdown, ScoreResult } from './types' -import type { MergedScoringInput } from './scoring-input' +import type { MergedScoringInput, UrlResolutionMap, UrlResolutionState } from './scoring-input' import { AUTHENTICITY_FAILURE_TIER, COMPLETENESS_WEIGHTS, FOUNDED_DATE_AGE_BUCKETS, SCORE_THRESHOLDS } from './constants' -import { resolvePublicUrl } from './link-resolver' import { evaluateMergedActorAuthenticity } from './scoring-authenticity' import { validateOrganizationLocationRef } from './location-utils' import { displayNameMatchesWebsiteDomain, normalizePublicWebsiteUrl } from './website-utils' @@ -26,6 +25,11 @@ const ZERO_BREAKDOWN: ScoreBreakdown = { banner: 0, } +// Organization records do not currently cap the number of URL refs in the +// generated lexicon. Keep URL scoring bounded and avoid network calls on the Tap +// ack path. +const MAX_ORGANIZATION_URLS_TO_CHECK = 3 + function normalizeText(value: unknown): string { return typeof value === 'string' ? value.trim() : '' } @@ -54,12 +58,17 @@ function scoreWebsitePresent(website: string | null): number { return website ? COMPLETENESS_WEIGHTS.websitePresent : 0 } -async function scoreWebsiteResolves(website: string | null): Promise { +function getUrlResolutionState(normalizedUrl: string, urlResolution?: UrlResolutionMap): UrlResolutionState { + return urlResolution?.[normalizedUrl] ?? 'unknown' +} + +function scoreWebsiteResolves(website: string | null, urlResolution?: UrlResolutionMap): number { const normalized = normalizePublicWebsiteUrl(website) if (!normalized) return 0 - const result = await resolvePublicUrl(normalized) - return result.resolvable ? COMPLETENESS_WEIGHTS.websiteResolves : 0 + return getUrlResolutionState(normalized, urlResolution) === 'failed' + ? 0 + : COMPLETENESS_WEIGHTS.websiteResolves } function scoreWebsiteMatchesName(displayName: string, website: string | null): number { @@ -71,16 +80,15 @@ function scoreOrganizationUrlsPresent(urls: MergedScoringInput['urls']): number return (urls ?? []).length > 0 ? COMPLETENESS_WEIGHTS.organizationUrlsPresent : 0 } -async function scoreOrganizationUrlsResolve(urls: MergedScoringInput['urls']): Promise { - for (const item of Array.isArray(urls) ? urls : []) { - const normalized = normalizePublicWebsiteUrl(item.url) - if (!normalized) continue +function scoreOrganizationUrlsResolve(urls: MergedScoringInput['urls'], urlResolution?: UrlResolutionMap): number { + const normalizedUrls = (Array.isArray(urls) ? urls : []) + .map(item => normalizePublicWebsiteUrl(item?.url)) + .filter((url): url is string => Boolean(url)) + .slice(0, MAX_ORGANIZATION_URLS_TO_CHECK) - const result = await resolvePublicUrl(normalized) - if (result.resolvable) return COMPLETENESS_WEIGHTS.organizationUrlsResolve - } + const hasResolvableUrl = normalizedUrls.some(url => getUrlResolutionState(url, urlResolution) !== 'failed') - return 0 + return hasResolvableUrl ? COMPLETENESS_WEIGHTS.organizationUrlsResolve : 0 } function scoreLocation(location: MergedScoringInput['location']): number { @@ -134,10 +142,8 @@ export async function scoreActivity(record: MergedScoringInput): Promise 0) return 'likely-test' if (score >= SCORE_THRESHOLDS['high-quality'].min) return 'high-quality' if (score >= SCORE_THRESHOLDS.standard.min) return 'standard' return 'likely-test' diff --git a/src/lib/scoring-authenticity.ts b/src/lib/scoring-authenticity.ts index a617f9b..eac3a15 100644 --- a/src/lib/scoring-authenticity.ts +++ b/src/lib/scoring-authenticity.ts @@ -1,6 +1,6 @@ import type { MergedScoringInput } from './scoring-input' -import { AUTHENTICITY_TEXT_PATTERNS } from './constants' -import { normalizePublicWebsiteUrl } from './website-utils' +import { AUTHENTICITY_TEXT_PATTERNS, DISPLAY_NAME_AUTHENTICITY_TEXT_PATTERNS } from './constants' +import { isPlaceholderWebsiteUrl, normalizePublicWebsiteUrl } from './website-utils' export interface AuthenticityGateResult { passed: boolean @@ -11,23 +11,39 @@ function normalizeText(value: unknown): string { return typeof value === 'string' ? value.trim().replace(/\s+/g, ' ') : '' } -function hasMeaningfulText(value: unknown): boolean { +function hasMeaningfulText(value: unknown, patterns: RegExp[] = AUTHENTICITY_TEXT_PATTERNS): boolean { const normalized = normalizeText(value) - return normalized.length > 0 && !AUTHENTICITY_TEXT_PATTERNS.some(pattern => pattern.test(normalized)) + return normalized.length > 0 && !patterns.some(pattern => pattern.test(normalized)) +} + +function hasRepeatedCharacterRun(value: string): boolean { + return /([\p{Letter}\p{Number}])\1{3,}/iu.test(value) } function pushSignal(signals: string[], signal: string): void { if (!signals.includes(signal)) signals.push(signal) } -function validateWebsite(value: string | null | undefined, signal: string, signals: string[]): boolean { +function validateWebsite( + value: string | null | undefined, + invalidUrlSignal: string, + placeholderDomainSignal: string, + signals: string[], +): boolean { const normalized = normalizeText(value) if (!normalized) return false - if (normalizePublicWebsiteUrl(normalized)) return true + if (!normalizePublicWebsiteUrl(normalized)) { + pushSignal(signals, invalidUrlSignal) + return false + } - pushSignal(signals, signal) - return false + if (isPlaceholderWebsiteUrl(normalized)) { + pushSignal(signals, placeholderDomainSignal) + return false + } + + return true } function validateFoundedDate(value: string | null | undefined, signals: string[]): boolean { @@ -54,6 +70,7 @@ function validateOrganizationUrls( ): { hasMeaningfulMetadata: boolean } { let hasMeaningfulMetadata = false let hasInvalidUrl = false + let hasPlaceholderDomain = false let hasPlaceholderLabel = false for (const item of urls ?? []) { @@ -61,10 +78,12 @@ function validateOrganizationUrls( const label = normalizeText(item.label) if (url) { - if (normalizePublicWebsiteUrl(url)) { - hasMeaningfulMetadata = true - } else { + if (!normalizePublicWebsiteUrl(url)) { hasInvalidUrl = true + } else if (isPlaceholderWebsiteUrl(url)) { + hasPlaceholderDomain = true + } else { + hasMeaningfulMetadata = true } } @@ -81,6 +100,10 @@ function validateOrganizationUrls( pushSignal(signals, 'Organization URL must be a public http(s) URL') } + if (hasPlaceholderDomain) { + pushSignal(signals, 'Organization URLs use placeholder domains') + } + if (hasPlaceholderLabel) { pushSignal(signals, 'Organization URL labels contain placeholder text') } @@ -92,10 +115,15 @@ export function evaluateMergedActorAuthenticity(record: MergedScoringInput): Aut const signals: string[] = [] const canonicalDisplayName = normalizeText(record.displayName) - if (canonicalDisplayName && !hasMeaningfulText(canonicalDisplayName)) { + const hasProfileDisplayName = record.displayNameSource !== 'did' + if (hasProfileDisplayName && canonicalDisplayName && !hasMeaningfulText(canonicalDisplayName, DISPLAY_NAME_AUTHENTICITY_TEXT_PATTERNS)) { pushSignal(signals, 'Display name contains placeholder text') } + if (hasProfileDisplayName && canonicalDisplayName && hasRepeatedCharacterRun(canonicalDisplayName)) { + pushSignal(signals, 'Display name contains repeated characters') + } + const profileDescription = normalizeText(record.profileDescription) if (profileDescription && !hasMeaningfulText(profileDescription)) { pushSignal(signals, 'Profile description contains placeholder text') @@ -109,14 +137,15 @@ export function evaluateMergedActorAuthenticity(record: MergedScoringInput): Aut const profileWebsite = validateWebsite( record.profileWebsite, 'Profile website must be a public http(s) URL', + 'Profile website uses placeholder domain', signals, ) const organizationUrls = validateOrganizationUrls(record.urls, signals) const foundedDate = validateFoundedDate(record.foundedDate, signals) - const displayNameIsMeaningful = record.displayNameSource !== 'did' && hasMeaningfulText(canonicalDisplayName) - const organizationTypeIsMeaningful = organizationTypeValues.some(hasMeaningfulText) + const displayNameIsMeaningful = hasProfileDisplayName && hasMeaningfulText(canonicalDisplayName, DISPLAY_NAME_AUTHENTICITY_TEXT_PATTERNS) + const organizationTypeIsMeaningful = organizationTypeValues.some(value => hasMeaningfulText(value)) const hasMeaningfulMetadata = displayNameIsMeaningful || diff --git a/src/lib/scoring-input.ts b/src/lib/scoring-input.ts index 2fae8dd..c6eeaff 100644 --- a/src/lib/scoring-input.ts +++ b/src/lib/scoring-input.ts @@ -8,6 +8,12 @@ export interface MergedOrganizationUrlItem { label: string | null } +/** Cached URL resolution state used by scoring without performing network I/O. */ +export type UrlResolutionState = 'unknown' | 'ok' | 'failed' + +/** Map of normalized public URL to its latest durable resolution state. */ +export type UrlResolutionMap = Record + export interface MergedScoringInput { did: string displayName: string @@ -20,6 +26,7 @@ export interface MergedScoringInput { hasBanner: boolean organizationType: string[] urls: MergedOrganizationUrlItem[] + urlResolution?: UrlResolutionMap location: OrganizationRecord['location'] | null foundedDate: string | null } diff --git a/src/lib/website-utils.ts b/src/lib/website-utils.ts index 522b1c2..a6c5ae4 100644 --- a/src/lib/website-utils.ts +++ b/src/lib/website-utils.ts @@ -1,4 +1,5 @@ import { isIP } from 'node:net' +import { PLACEHOLDER_DOMAINS, PLACEHOLDER_TLDS } from './constants' const COMMON_2ND_LEVEL_SUFFIXES = new Set([ 'ac', @@ -82,6 +83,17 @@ function isPrivateIpv6(hostname: string): boolean { ) } +/** Returns true when a literal IP address is globally routable enough for URL enrichment fetches. */ +export function isPublicNetworkAddress(address: string): boolean { + const normalized = address.replace(/^\[(.*)\]$/, '$1').toLowerCase() + const ipVersion = isIP(normalized) + + if (ipVersion === 4) return !isPrivateIpv4(normalized) + if (ipVersion === 6) return !isPrivateIpv6(normalized) + + return false +} + function isPublicHostname(hostname: string): boolean { const normalized = normalizeHostname(hostname) @@ -90,8 +102,7 @@ function isPublicHostname(hostname: string): boolean { } const ipVersion = isIP(normalized) - if (ipVersion === 4) return !isPrivateIpv4(normalized) - if (ipVersion === 6) return !isPrivateIpv6(normalized) + if (ipVersion === 4 || ipVersion === 6) return isPublicNetworkAddress(normalized) if (normalized.startsWith('.') || normalized.endsWith('.') || normalized.includes('..')) return false if (/^[0-9.]+$/.test(normalized)) return false @@ -157,6 +168,26 @@ export function normalizePublicWebsiteUrl(value: string | null | undefined): str return parsed.toString() } +function isPlaceholderHostname(hostname: string): boolean { + const normalized = normalizeHostname(hostname) + const labels = normalized.split('.').filter(Boolean) + const tld = labels.at(-1) + + if (PLACEHOLDER_DOMAINS.some(domain => normalized === domain || normalized.endsWith(`.${domain}`))) { + return true + } + + return Boolean(tld && PLACEHOLDER_TLDS.includes(tld as typeof PLACEHOLDER_TLDS[number])) +} + +/** Returns true when a syntactically valid public URL points at a reserved placeholder host. */ +export function isPlaceholderWebsiteUrl(value: string | null | undefined): boolean { + const normalizedUrl = normalizePublicWebsiteUrl(value) + if (!normalizedUrl) return false + + return isPlaceholderHostname(new URL(normalizedUrl).hostname) +} + export function getWebsiteDomainStem(value: string | null | undefined): string | null { const normalizedUrl = normalizePublicWebsiteUrl(value) if (!normalizedUrl) return null diff --git a/tests/pds-test-override.test.ts b/tests/pds-test-override.test.ts new file mode 100644 index 0000000..56f75e2 --- /dev/null +++ b/tests/pds-test-override.test.ts @@ -0,0 +1,49 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { applyPdsTestOverride, updatePdsTestSignals } from '../src/lib/pds-test-override' +import type { ScoreResult } from '../src/lib/types' + +function scoreResult(): ScoreResult { + return { + totalScore: 90, + tier: 'high-quality', + breakdown: { + displayName: 20, + description: 15, + organizationType: 10, + websitePresent: 10, + websiteResolves: 10, + websiteMatchesName: 5, + organizationUrlsPresent: 5, + organizationUrlsResolve: 5, + locationValid: 0, + foundedDateValid: 5, + foundedDateAge: 0, + avatar: 5, + banner: 0, + }, + testSignals: [], + } +} + +test('configured actor PDS host forces likely-test without changing score', () => { + const result = applyPdsTestOverride( + scoreResult(), + 'https://EPDS1.test.certified.app/', + ['epds1.test.certified.app'], + ) + + assert.equal(result.totalScore, 90) + assert.equal(result.tier, 'likely-test') + assert.deepEqual(result.testSignals, ['actor-pds-test-host: epds1.test.certified.app']) +}) + +test('stale actor PDS signals are removed when host is no longer configured', () => { + const result = updatePdsTestSignals( + ['actor-pds-test-host: epds1.test.certified.app', 'Display name contains placeholder text'], + 'bsky.social', + ['epds1.test.certified.app'], + ) + + assert.deepEqual(result, ['Display name contains placeholder text']) +}) diff --git a/tests/pending-organization-deletes.test.ts b/tests/pending-organization-deletes.test.ts new file mode 100644 index 0000000..a0fcfd5 --- /dev/null +++ b/tests/pending-organization-deletes.test.ts @@ -0,0 +1,85 @@ +import assert from 'node:assert/strict' +import { after, beforeEach, test } from 'node:test' +import { mkdtempSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import type { OrganizationSnapshot } from '../src/lib/types' + +const testDir = mkdtempSync(join(tmpdir(), 'orglabeler-pending-deletes-')) +process.env.ACTIVITY_DB_PATH = join(testDir, 'activity-log.db') +process.env.LABELS_DB_PATH = join(testDir, 'labels.db') + +const { + closeDb, + completePendingOrganizationDelete, + deletePendingOrganizationDelete, + getDb, + getOrganizationSnapshot, + hasPendingOrganizationDelete, + upsertOrganizationSnapshot, + upsertPendingOrganizationDelete, +} = await import('../src/lib/db') + +beforeEach(() => { + getDb().exec(` + DELETE FROM activities; + DELETE FROM organization_snapshots; + DELETE FROM pending_organization_deletes; + `) +}) + +after(() => { + closeDb() + rmSync(testDir, { recursive: true, force: true }) +}) + +function organizationPayload(): OrganizationSnapshot['payload'] { + return { + $type: 'app.certified.actor.organization', + organizationType: ['ngo'], + createdAt: '2024-01-01T00:00:00.000Z', + } +} + +test('completing a pending delete removes only the matching local organization state', () => { + const did = 'did:example:delete' + const rkey = 'self' + const recordUri = `at://${did}/app.certified.actor.organization/${rkey}` + + upsertOrganizationSnapshot({ + did, + rkey, + recordUri, + payload: organizationPayload(), + }) + upsertPendingOrganizationDelete(did, rkey, recordUri) + + assert.equal(completePendingOrganizationDelete(did, rkey, recordUri), true) + assert.equal(getOrganizationSnapshot(did), null) + assert.equal(hasPendingOrganizationDelete(did), false) +}) + +test('stale pending delete cleanup does not remove a newer replacement snapshot', () => { + const did = 'did:example:delete-race' + const staleRkey = 'old' + const staleUri = `at://${did}/app.certified.actor.organization/${staleRkey}` + const freshRkey = 'self' + const freshUri = `at://${did}/app.certified.actor.organization/${freshRkey}` + + upsertPendingOrganizationDelete(did, staleRkey, staleUri) + + deletePendingOrganizationDelete(did) + upsertOrganizationSnapshot({ + did, + rkey: freshRkey, + recordUri: freshUri, + payload: organizationPayload(), + }) + + assert.equal(completePendingOrganizationDelete(did, staleRkey, staleUri), false) + + const current = getOrganizationSnapshot(did) + assert.ok(current) + assert.equal(current.rkey, freshRkey) + assert.equal(current.recordUri, freshUri) +}) diff --git a/tests/recompute-jobs.test.ts b/tests/recompute-jobs.test.ts new file mode 100644 index 0000000..a58fcf8 --- /dev/null +++ b/tests/recompute-jobs.test.ts @@ -0,0 +1,79 @@ +import assert from 'node:assert/strict' +import { after, test } from 'node:test' +import { mkdtempSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { tmpdir } from 'node:os' + +const testDir = mkdtempSync(join(tmpdir(), 'orglabeler-recompute-jobs-')) +process.env.ACTIVITY_DB_PATH = join(testDir, 'activity-log.db') +process.env.LABELS_DB_PATH = join(testDir, 'labels.db') + +const { + claimDueRecomputeJob, + closeDb, + completeRecomputeJob, + enqueueRecomputeJob, + failRecomputeJob, + getDb, + getRecomputeJobCounts, + recoverStaleRunningRecomputeJobs, +} = await import('../src/lib/db') + +after(() => { + closeDb() + rmSync(testDir, { recursive: true, force: true }) +}) + +function dueNow(): string { + return new Date(Date.now() + 1000).toISOString() +} + +test('recompute jobs coalesce by DID and preserve newer pending work during races', () => { + enqueueRecomputeJob('recompute-org', 'did:example:one', { delayMs: 50, payload: { reason: 'first' } }) + enqueueRecomputeJob('recompute-org', 'did:example:one', { delayMs: 0, payload: { reason: 'second' } }) + + assert.equal(getRecomputeJobCounts().pending, 1) + + let job = claimDueRecomputeJob(dueNow()) + assert.ok(job) + assert.equal(job.attempts, 1) + + enqueueRecomputeJob('recompute-org', 'did:example:one', { delayMs: 0, payload: { reason: 'raced' } }) + assert.equal(completeRecomputeJob(job.id), 0) + + job = claimDueRecomputeJob(dueNow()) + assert.ok(job) + assert.equal(completeRecomputeJob(job.id), 1) + + enqueueRecomputeJob('recompute-org', 'did:example:two', { delayMs: 0 }) + job = claimDueRecomputeJob(dueNow()) + assert.ok(job) + + enqueueRecomputeJob('recompute-org', 'did:example:two', { delayMs: 0, payload: { reason: 'retry-race' } }) + failRecomputeJob(job.id, 'old worker failed', 60_000) + + job = claimDueRecomputeJob(dueNow()) + assert.ok(job) + assert.equal(completeRecomputeJob(job.id), 1) + + const counts = getRecomputeJobCounts() + assert.equal(counts.pending, 0) + assert.equal(counts.running, 0) + assert.equal(counts.failed, 0) + assert.equal(counts.done, 2) +}) + +test('stale running recompute jobs are recovered for retry', () => { + enqueueRecomputeJob('recompute-org', 'did:example:stale', { delayMs: 0 }) + + let job = claimDueRecomputeJob(dueNow()) + assert.ok(job) + + getDb().prepare("UPDATE recompute_jobs SET updated_at = datetime('now', '-10 minutes') WHERE id = ?").run(job.id) + + assert.equal(recoverStaleRunningRecomputeJobs(5 * 60_000), 1) + + job = claimDueRecomputeJob(dueNow()) + assert.ok(job) + assert.equal(completeRecomputeJob(job.id), 1) +}) diff --git a/tests/scoring-authenticity.test.ts b/tests/scoring-authenticity.test.ts new file mode 100644 index 0000000..66e0285 --- /dev/null +++ b/tests/scoring-authenticity.test.ts @@ -0,0 +1,145 @@ +import assert from 'node:assert/strict' +import { test } from 'node:test' +import { evaluateMergedActorAuthenticity } from '../src/lib/scoring-authenticity' +import type { MergedScoringInput } from '../src/lib/scoring-input' + +function record(overrides: Partial = {}): MergedScoringInput { + return { + did: 'did:plc:exampleauthenticity', + displayName: 'Forest Recovery Collective', + displayNameSource: 'profile', + profileDisplayName: 'Forest Recovery Collective', + profileDescription: 'A community organization restoring native forest habitats.', + profileWebsite: null, + validationNotes: [], + hasAvatar: true, + hasBanner: false, + organizationType: ['nonprofit'], + urls: [{ url: 'https://forest-recovery.example.coop', label: 'Website' }], + location: null, + foundedDate: null, + ...overrides, + } +} + +test('display-name workflow and test terms fail the authenticity gate', () => { + const names = [ + 'HC Community Demo', + 'Ma Earth (Dev)', + 'Ma Earth Staging', + 'Alpha QA Org', + 'Staging E2E Org mn8y027w', + 'Sandbox Org', + 'Fixture Org', + 'tobytest', + 'exclusivecgstester1', + 'Seed Data Org', + 'Org demo new DB', + 'Published', + 'Unpublished org', + 'CHANGES REQUESTED ORG', + 'ORG', + 'ORG 1', + 'First org', + 'New Org', + ] + + for (const displayName of names) { + const result = evaluateMergedActorAuthenticity(record({ displayName, profileDisplayName: displayName })) + assert.equal(result.passed, false, displayName) + assert.ok(result.signals.includes('Display name contains placeholder text'), displayName) + } +}) + +test('display-name workflow checks do not punish longer descriptive words or non-display fields', () => { + const result = evaluateMergedActorAuthenticity(record({ + displayName: 'Community Development Alliance', + profileDisplayName: 'Community Development Alliance', + profileDescription: 'We support development teams with demo plots, staging areas, QA review, and E2E checks for local cooperatives.', + })) + + assert.equal(result.passed, true) + assert.deepEqual(result.signals, []) +}) + +test('display-name embedded-test checks do not punish normal words', () => { + const names = [ + 'Contest Foundation', + 'Protest Relief Network', + 'Latest Research Cooperative', + 'Attest Community Trust', + 'Detest River Cleanup', + ] + + for (const displayName of names) { + const result = evaluateMergedActorAuthenticity(record({ displayName, profileDisplayName: displayName })) + assert.equal(result.passed, true, displayName) + assert.deepEqual(result.signals, [], displayName) + } +}) + +test('placeholder profile website domains fail authenticity checks', () => { + const urls = [ + 'https://example.com/about', + 'https://example.net/about', + 'https://example.org/about', + 'https://registry.example/actor', + 'https://registry.invalid/actor', + 'https://registry.test/actor', + ] + + for (const profileWebsite of urls) { + const result = evaluateMergedActorAuthenticity(record({ profileWebsite })) + assert.equal(result.passed, false, profileWebsite) + assert.ok(result.signals.includes('Profile website uses placeholder domain'), profileWebsite) + } +}) + +test('placeholder organization URL domains fail authenticity checks', () => { + const urls = [ + 'https://docs.example.com/project', + 'https://docs.example.net/project', + 'https://docs.example.org/project', + 'https://registry.example/actor', + 'https://registry.invalid/actor', + 'https://registry.test/actor', + ] + + for (const url of urls) { + const result = evaluateMergedActorAuthenticity(record({ + urls: [{ url, label: 'Project docs' }], + })) + assert.equal(result.passed, false, url) + assert.ok(result.signals.includes('Organization URLs use placeholder domains'), url) + } +}) + +test('DID fallback display names are not checked as user-provided display names', () => { + const result = evaluateMergedActorAuthenticity(record({ + displayName: 'aaaa12345678…', + displayNameSource: 'did', + profileDisplayName: null, + })) + + assert.equal(result.passed, true) + assert.deepEqual(result.signals, []) +}) + +test('display-name repeated character runs fail authenticity checks', () => { + const result = evaluateMergedActorAuthenticity(record({ + displayName: 'DripsOrrrrr', + profileDisplayName: 'DripsOrrrrr', + })) + + assert.equal(result.passed, false) + assert.ok(result.signals.includes('Display name contains repeated characters')) +}) + +test('lorem ipsum anywhere in a description fails authenticity checks', () => { + const result = evaluateMergedActorAuthenticity(record({ + profileDescription: 'What is Lorem Ipsum? Lorem Ipsum is simply dummy text.', + })) + + assert.equal(result.passed, false) + assert.ok(result.signals.includes('Profile description contains placeholder text')) +}) diff --git a/tests/url-check-cache.test.ts b/tests/url-check-cache.test.ts new file mode 100644 index 0000000..04e7a87 --- /dev/null +++ b/tests/url-check-cache.test.ts @@ -0,0 +1,108 @@ +import assert from 'node:assert/strict' +import { after, beforeEach, test } from 'node:test' +import { mkdtempSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { scoreActivity } from '../src/lib/scorer' +import type { MergedScoringInput } from '../src/lib/scoring-input' + +const testDir = mkdtempSync(join(tmpdir(), 'orglabeler-url-checks-')) +process.env.ACTIVITY_DB_PATH = join(testDir, 'activity-log.db') +process.env.LABELS_DB_PATH = join(testDir, 'labels.db') + +const { + closeDb, + getDb, + getDueUrlCheck, + getUrlResolutionMap, + recordUrlCheckFailure, + recordUrlCheckOk, + upsertPendingUrlCheck, +} = await import('../src/lib/db') + +beforeEach(() => { + getDb().exec('DELETE FROM url_checks') +}) + +after(() => { + closeDb() + rmSync(testDir, { recursive: true, force: true }) +}) + +test('rediscovery preserves pending URL attempts so hard failures can converge', () => { + const url = 'https://missing.example.org/' + + upsertPendingUrlCheck(url) + recordUrlCheckFailure({ + normalizedUrl: url, + status: 'pending', + resolvable: null, + statusCode: 404, + error: 'HTTP 404', + attempts: 1, + retryAfterMs: -1000, + }) + + upsertPendingUrlCheck(url) + + const due = getDueUrlCheck() + assert.ok(due) + assert.equal(due.normalizedUrl, url) + assert.equal(due.status, 'pending') + assert.equal(due.attempts, 1) + assert.equal(due.statusCode, 404) +}) + +test('expired ok URL checks are refreshed as new pending checks', () => { + const url = 'https://stale.example.org/' + + upsertPendingUrlCheck(url) + recordUrlCheckOk(url, 200, -1000) + upsertPendingUrlCheck(url) + + const due = getDueUrlCheck() + assert.ok(due) + assert.equal(due.normalizedUrl, url) + assert.equal(due.status, 'pending') + assert.equal(due.attempts, 0) + assert.equal(due.statusCode, null) +}) + +test('failed URL cache state removes provisional URL resolve points', async () => { + const url = 'https://forest-recovery.example.coop/' + const base: MergedScoringInput = { + did: 'did:example:score', + displayName: 'Example Org', + displayNameSource: 'profile', + profileDisplayName: 'Example Org', + profileDescription: 'A real organization', + profileWebsite: url, + validationNotes: [], + hasAvatar: false, + hasBanner: false, + organizationType: ['ngo'], + urls: [{ url, label: 'Website' }], + location: null, + foundedDate: null, + } + + upsertPendingUrlCheck(url) + assert.deepEqual(getUrlResolutionMap([url]), {}) + + const optimistic = await scoreActivity(base) + + recordUrlCheckFailure({ + normalizedUrl: url, + status: 'failed', + resolvable: false, + statusCode: 404, + error: 'HTTP 404', + attempts: 2, + retryAfterMs: 60_000, + }) + + assert.deepEqual(getUrlResolutionMap([url]), { [url]: 'failed' }) + + const failed = await scoreActivity({ ...base, urlResolution: { [url]: 'failed' } }) + assert.ok(optimistic.totalScore > failed.totalScore) +}) diff --git a/tests/url-enrichment-pds-gate.test.ts b/tests/url-enrichment-pds-gate.test.ts new file mode 100644 index 0000000..f8bd869 --- /dev/null +++ b/tests/url-enrichment-pds-gate.test.ts @@ -0,0 +1,91 @@ +import assert from 'node:assert/strict' +import { after, beforeEach, test } from 'node:test' +import { mkdtempSync, rmSync } from 'node:fs' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import type { OrganizationSnapshot } from '../src/lib/types' + +const testDir = mkdtempSync(join(tmpdir(), 'orglabeler-url-pds-gate-')) +process.env.ACTIVITY_DB_PATH = join(testDir, 'activity-log.db') +process.env.LABELS_DB_PATH = join(testDir, 'labels.db') +process.env.TEST_PDS_HOSTS = 'epds1.test.certified.app' + +const { + closeDb, + getDb, + getDueUrlCheck, + getRecomputeJobCounts, + recordActorPdsOk, + upsertOrganizationSnapshot, +} = await import('../src/lib/db') +const { enqueueUrlChecksForDid } = await import('../src/labeler/url-enrichment-worker') + +after(() => { + closeDb() + rmSync(testDir, { recursive: true, force: true }) +}) + +beforeEach(() => { + getDb().exec(` + DELETE FROM actor_pds_cache; + DELETE FROM recompute_jobs; + DELETE FROM url_checks; + DELETE FROM organization_snapshots; + DELETE FROM profile_snapshots; + `) +}) + +function upsertOrganization(did: string): void { + upsertOrganizationSnapshot({ + did, + rkey: 'self', + recordUri: `at://${did}/app.certified.actor.organization/self`, + payload: { + $type: 'app.certified.actor.organization', + organizationType: ['ngo'], + urls: [{ url: 'https://forest-recovery.example.coop', label: 'Website' }], + createdAt: '2024-01-01T00:00:00.000Z', + } as OrganizationSnapshot['payload'], + }) +} + +test('URL enrichment is deferred until actor PDS cache is known when TEST_PDS_HOSTS is configured', () => { + const did = 'did:plc:unknownpds' + upsertOrganization(did) + + assert.equal(enqueueUrlChecksForDid(did), 0) + assert.equal(getDueUrlCheck(), null) + assert.equal(getRecomputeJobCounts().pending, 1) +}) + +test('URL enrichment is skipped for actors on configured test PDS hosts', () => { + const did = 'did:plc:testpds' + upsertOrganization(did) + recordActorPdsOk(did, 'https://epds1.test.certified.app', 'epds1.test.certified.app', 60_000) + + assert.equal(enqueueUrlChecksForDid(did), 0) + assert.equal(getDueUrlCheck(), null) + assert.equal(getRecomputeJobCounts().pending, 0) +}) + +test('URL enrichment runs for actors on fresh non-test PDS hosts', () => { + const did = 'did:plc:prod' + upsertOrganization(did) + recordActorPdsOk(did, 'https://bsky.social', 'bsky.social', 60_000) + + assert.equal(enqueueUrlChecksForDid(did), 1) + + const due = getDueUrlCheck() + assert.ok(due) + assert.equal(due.normalizedUrl, 'https://forest-recovery.example.coop/') +}) + +test('URL enrichment is deferred for stale PDS cache rows', () => { + const did = 'did:plc:stalepds' + upsertOrganization(did) + recordActorPdsOk(did, 'https://bsky.social', 'bsky.social', -1000) + + assert.equal(enqueueUrlChecksForDid(did), 0) + assert.equal(getDueUrlCheck(), null) + assert.equal(getRecomputeJobCounts().pending, 1) +})