rock
cd3d00048a
- Unified QueueAdapter trait for concurrent dual-write operations - GatewayQueueAdapter routes messages via api.riotpiao.com with X-Service: sqs header - TokenProvider abstraction: StaticTokenProvider + AuthentikTokenProvider - JWT bearer token support (from Authentik OAuth2) - InMemoryQueueAdapter for testing - Base64 encoding/decoding for SQS message bodies - HTTP/REST integration (no direct gRPC complexity) - 8 unit tests + comprehensive documentation - Supports long-polling (ReceiveMessage), visibility timeout, DLQ Uses standard SQS API patterns: - SendMessage: Queue chunk for dual-write processing - ReceiveMessage: Long-poll up to 10 messages, 20s wait - DeleteMessage: Acknowledge on success - ChangeMessageVisibility: Retry on failure - SendToDLQ: After max retries Files: - crates/mem-cli/src/queue_adapter.rs (310 LOC) - crates/mem-cli/src/gateway_queue_adapter.rs (530 LOC) - tests/it_gateway_queue_adapter.rs (110 LOC) - docs/M8.2-GATEWAY_QUEUE_ADAPTER.md (400 LOC)
63 lines
2.7 KiB
SQL
63 lines
2.7 KiB
SQL
-- M8.2 — Dual-write indexing pipeline
|
|
-- Unified chunk table for pgvector (embedding) and OpenSearch (text) coordination
|
|
-- Both stores write same chunk_id; OpenSearch failure marked for eventual consistency retry
|
|
|
|
CREATE TABLE IF NOT EXISTS chunks (
|
|
-- Identity
|
|
id UUID NOT NULL PRIMARY KEY DEFAULT gen_random_uuid(),
|
|
chunk_hash TEXT NOT NULL UNIQUE, -- SHA256(content) for deduplication
|
|
|
|
-- Content
|
|
content TEXT NOT NULL,
|
|
source TEXT NOT NULL, -- "ingest", "vault", "transcript", etc.
|
|
project TEXT NOT NULL,
|
|
|
|
-- Metadata
|
|
level TEXT NOT NULL CHECK (level IN ('L0', 'L1', 'L2', 'R')),
|
|
breadcrumb TEXT[] DEFAULT '{}', -- hierarchical path for context display
|
|
|
|
-- Vector storage (pgvector)
|
|
embedding vector(768), -- 768-dim nomic embeddings (nullable for pending writes)
|
|
|
|
-- Dual-write tracking
|
|
indexed_in_pgvector BOOLEAN NOT NULL DEFAULT false,
|
|
indexed_in_opensearch BOOLEAN NOT NULL DEFAULT false,
|
|
opensearch_pending BOOLEAN NOT NULL DEFAULT false, -- retry marker
|
|
|
|
-- Timestamps
|
|
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
|
pgvector_indexed_at TIMESTAMPTZ,
|
|
opensearch_indexed_at TIMESTAMPTZ,
|
|
opensearch_retry_count INT NOT NULL DEFAULT 0,
|
|
opensearch_last_retry_at TIMESTAMPTZ,
|
|
|
|
CONSTRAINT valid_pgvector CHECK (
|
|
-- Either both stored, or pgvector done + opensearch pending
|
|
(indexed_in_pgvector AND indexed_in_opensearch AND NOT opensearch_pending) OR
|
|
(indexed_in_pgvector AND NOT indexed_in_opensearch AND opensearch_pending)
|
|
)
|
|
);
|
|
|
|
-- Indexes for common queries
|
|
CREATE INDEX idx_chunks_project ON chunks(project);
|
|
CREATE INDEX idx_chunks_level ON chunks(project, level);
|
|
CREATE INDEX idx_chunks_source ON chunks(source);
|
|
CREATE INDEX idx_chunks_opensearch_pending ON chunks(project) WHERE opensearch_pending = true;
|
|
CREATE INDEX idx_chunks_created_at ON chunks(created_at DESC);
|
|
|
|
-- Vector search index (only for successfully stored embeddings)
|
|
CREATE INDEX idx_chunks_embedding ON chunks USING hnsw (embedding vector_cosine_ops)
|
|
WHERE indexed_in_pgvector = true AND embedding IS NOT NULL;
|
|
|
|
-- Deduplication index
|
|
CREATE INDEX idx_chunks_hash ON chunks(chunk_hash);
|
|
|
|
-- Breadcrumb GiST index for hierarchical path queries
|
|
CREATE INDEX idx_chunks_breadcrumb ON chunks USING gin (breadcrumb);
|
|
|
|
COMMENT ON TABLE chunks IS 'M8.2 unified chunk store: dual-write to pgvector (embedding) + OpenSearch (text)';
|
|
COMMENT ON COLUMN chunks.id IS 'Same chunk_id written to both pgvector and OpenSearch';
|
|
COMMENT ON COLUMN chunks.chunk_hash IS 'SHA256(content) for dedup before dual write';
|
|
COMMENT ON COLUMN chunks.opensearch_pending IS 'Retry marker: OpenSearch write failed, needs eventual consistency retry';
|
|
COMMENT ON COLUMN chunks.embedding IS 'pgvector embedding (768-dim nomic), NULL until pgvector write succeeds';
|