Finish the hexagonal ports-&-adapters split started in #183. The producer side
already had a TaskProducer port + adapters, but the consumer side was
unabstracted and the INGEST_QUEUE selection leaked into a duplicated
`if use_postgres:` branch across both app.py lifespan paths.
Introduce an IngestTransport ABC (vector/queue/transport.py) that bundles the
producer with running (or not running) the in-process consumer pool, built by a
single build_transport() factory:
- LocalTransport (INGEST_QUEUE=memory): in-process anyio stream drained by an
N-worker pool that run_consumers starts.
- DistributedTransport (INGEST_QUEUE=postgres): wraps ProcrastinateTaskProducer;
run_consumers is a no-op because the consumer is the external `worker` role.
Both lifespan paths now call build_transport + _wire_vector_sync_state (new
helper that centralizes the app.state / module-singleton / browser-app writes) +
transport.run_consumers + transport.aclose(), with no INGEST_QUEUE branching and
no getattr drain probe. Adding a future backend (Redis/NATS/SQS) is one new
adapter + one build_transport arm, with no app.py or scanner change.
Preserves the single-tenant parallelism invariant (one shared multiplexed queue
+ N-worker pool, per-document not per-user dispatch) and documents it in
ADR-028. The worker CLI is unchanged (it is the external consumer).
Refs: Deck #196 (Deck #197 tracks the explicit parallelism regression test)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Re-architect document ingest from the shared NATS-glued document-processor to a
per-tenant, in-process model owned by nextcloud-mcp-server (Deck #183). The MCP
server now owns both sides of ingest:
- Producer (api role): the scanner defers one job per changed document into the
app's Postgres via procrastinate (queueing_lock dedup; no execution lock, so a
crashed worker can't deadlock a doc — Qdrant upserts are idempotent).
- Consumer (worker role): `nextcloud-mcp-server worker` drains the queue and runs
the existing process_document pipeline; a periodic task reclaims jobs orphaned
in `doing` by a crash.
INGEST_QUEUE selects the transport (auto: postgres when DATABASE_URL is Postgres,
else the in-process anyio queue for SQLite/dev). procrastinate manages its own
tables (applied on a fresh DB at startup and by `db upgrade`). The vector-sync
status surface reads job counts from Postgres in postgres mode. procrastinate +
psycopg3 ship in the [postgres] extra; the app's own engine still uses asyncpg
(driver unification is a follow-up handled in the rendered Helm chart).
NATS JetStream, the Postgres-queue stub, the bus status subscriber, and nats-py
are removed.
BREAKING CHANGE: the external-NATS-ingest env vars are removed
(INGEST_MODE, STATUS_BACKEND, INGEST_BUS_URL, INGEST_BUS_NUM_REPLICAS,
FACT_EVENT_EMITTER). Use INGEST_QUEUE (memory|postgres) and the `worker`
command instead. TENANT_ID is retained (no longer NATS-subject-charset-validated).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Adds the seven §10.2 hook-point modules + five env vars so Astrolabe Cloud can
offload document processing to the external document-processor / embedding
gateway. Purely additive: with every setting unset the server behaves exactly
as today, so self-hosters are unaffected (Deck #92).
Hook points (all default to current monolith behavior):
- config: EMBEDDING_PROVIDER, INGEST_MODE, STATUS_BACKEND,
COLLECTION_METADATA_SOURCE, FACT_EVENT_EMITTER (+ supporting settings),
validated in Settings.__post_init__ (fail-fast STATUS_BACKEND=local with
INGEST_MODE=external); shared canonical.py.
- vector/payload_keys.py + acl_hash.py: cross-impl NAMESPACE/point_id (§2.2)
and BLAKE2b-128 ACL hash (§11), pinned by fixtures shared with the
document-processor repo.
- embedding/gateway_client.py: OpenAI-compatible GatewayProvider authenticating
via M2M OIDC client-credentials (separate realm); manual-only registry entry.
- vector/collection_metadata.py: sentinel-point / API metadata source with env
fallback.
- vector/queue/: hexagonal ingest producer ports + memory/NATS adapters
(Postgres seam); INGEST_MODE=external publishes mcp.ingest.requested.{tenant}
instead of the in-memory stream and skips the in-process processor pool. The
lifespan becomes a composition root across both deployment branches.
- vector/queue/status.py: STATUS_BACKEND=bus subscriber feeding a StatusStore
the vector-sync status endpoint reads.
- admin/payload_backfill.py: POST /api/v1/admin/payload-backfill (admin scope);
processor writes the new payload keys; query-side ACL pre-filter gated behind
ACL_PREFILTER_ENABLED (default off).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
