From 2695b5d91e120e76bdd9a5ec241dac8358e12b73 Mon Sep 17 00:00:00 2001 From: Mathias Date: Tue, 2 Jun 2026 20:04:04 +0200 Subject: [PATCH] feat(adapters): add Postgres store sink with durable dedup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements ports.Sink over Postgres (pgx/v5 + pgxpool, DSN from env per estate convention). This is the primary sink (ADR-003) and the source of the engine's durable, cross-restart dedup — the in-engine processed map is process-lifetime only. - Migrations (golang-migrate, NNN_name.up/down.sql per estate convention, applied from an embedded FS on New): users, videos, transcripts, summaries, sink_deliveries. Every user-owned table carries user_id (Stage-0 per-user isolation promise, data-model.md). summaries has UNIQUE(user_id, video_id) — at most one summary per video; highlights / takeaways are jsonb. - Deliver upserts the summary idempotently on (user_id, video_id) (ON CONFLICT DO UPDATE) inside one tx with its sink_delivery row. Re- delivering the same summary updates in place, never duplicates or errors. - Dedup reads (store methods, not a new port): HasSummary(ctx,userID, videoID) and SeenVideoIDs(ctx,userID) — both user_id-scoped, so one user never sees another's videos. summaries.video_id is intentionally not FK-constrained to videos at Stage 0: the sink receives only a Summary, so the dedup key stands alone; video-row persistence is the engine/source's concern, deferred. Tested against a real in-process Postgres via embedded-postgres (real SQL: constraints, ON CONFLICT, jsonb, user_id scoping) — no docker, no live cluster, no creds, fully offline. Deps: golang-migrate/migrate/v4 and jackc/pgx/v5 (runtime), fergusstrange/embedded-postgres + stretchr/testify (test-only). go mod tidy raised the go directive to 1.25.0 (minimum required by the dep graph; estate elsewhere already runs 1.26.1). Co-Authored-By: Claude Opus 4.8 (1M context) --- go.mod | 25 ++- go.sum | 97 +++++++++ .../store/migrations/001_initial.down.sql | 5 + .../store/migrations/001_initial.up.sql | 73 +++++++ internal/adapters/store/store.go | 201 ++++++++++++++++++ internal/adapters/store/store_test.go | 173 +++++++++++++++ 6 files changed, 572 insertions(+), 2 deletions(-) create mode 100644 internal/adapters/store/migrations/001_initial.down.sql create mode 100644 internal/adapters/store/migrations/001_initial.up.sql create mode 100644 internal/adapters/store/store.go create mode 100644 internal/adapters/store/store_test.go diff --git a/go.mod b/go.mod index 68341d2..f798053 100644 --- a/go.mod +++ b/go.mod @@ -1,5 +1,26 @@ module gitea.d-ma.be/mathias/tapir -go 1.23.0 +go 1.25.0 -require golang.org/x/oauth2 v0.30.0 +require ( + github.com/fergusstrange/embedded-postgres v1.34.0 + github.com/golang-migrate/migrate/v4 v4.19.1 + github.com/jackc/pgx/v5 v5.9.2 + github.com/stretchr/testify v1.11.1 + golang.org/x/oauth2 v0.30.0 +) + +require ( + github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa // indirect + github.com/jackc/pgpassfile v1.0.0 // indirect + github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect + github.com/jackc/puddle/v2 v2.2.2 // indirect + github.com/lib/pq v1.10.9 // indirect + github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect + github.com/rogpeppe/go-internal v1.15.0 // indirect + github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect + golang.org/x/sync v0.18.0 // indirect + golang.org/x/text v0.31.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/go.sum b/go.sum index c05b468..c950b79 100644 --- a/go.sum +++ b/go.sum @@ -1,2 +1,99 @@ +github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 h1:L/gRVlceqvL25UVaW/CKtUDjefjrs0SPonmDGUVOYP0= +github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= +github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= +github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= +github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= +github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= +github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dhui/dktest v0.4.6 h1:+DPKyScKSEp3VLtbMDHcUq6V5Lm5zfZZVb0Sk7Ahom4= +github.com/dhui/dktest v0.4.6/go.mod h1:JHTSYDtKkvFNFHJKqCzVzqXecyv+tKt8EzceOmQOgbU= +github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk= +github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E= +github.com/docker/docker v28.3.3+incompatible h1:Dypm25kh4rmk49v1eiVbsAtpAsYURjYkaKubwuBdxEI= +github.com/docker/docker v28.3.3+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk= +github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c= +github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc= +github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4= +github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk= +github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg= +github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U= +github.com/fergusstrange/embedded-postgres v1.34.0 h1:c6RKhPKFsLVU+Tdxsx8q0UxCHsvZZ/iShAnljRBXs6s= +github.com/fergusstrange/embedded-postgres v1.34.0/go.mod h1:w0YvnCgf19o6tskInrOOACtnqfVlOvluz3hlNLY7tRk= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang-migrate/migrate/v4 v4.19.1 h1:OCyb44lFuQfYXYLx1SCxPZQGU7mcaZ7gH9yH4jSFbBA= +github.com/golang-migrate/migrate/v4 v4.19.1/go.mod h1:CTcgfjxhaUtsLipnLoQRWCrjYXycRz/g5+RWDuYgPrE= +github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa h1:s+4MhCQ6YrzisK6hFJUX53drDT4UsSW3DEhKn0ifuHw= +github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds= +github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM= +github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo= +github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM= +github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw= +github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= +github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= +github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= +github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= +github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw= +github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o= +github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= +github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= +github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0= +github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y= +github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A= +github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc= +github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= +github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= +github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug= +github.com/opencontainers/image-spec v1.1.0/go.mod h1:W4s4sFTMaBeK1BQLXbG4AdM2szdn85PY75RI83NrTrM= +github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.15.0 h1:D0RCU5rMAp+SpgkiNdrjfJ+LX4J1M32V2NeCY7EJ6hc= +github.com/rogpeppe/go-internal v1.15.0/go.mod h1:DrUVZyrJU+txYW5/1kwtXQSMFio52ZOxX7yM1VHvnxs= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 h1:nIPpBwaJSVYIxUFsDv3M8ofmx9yWTog9BfvIu0q41lo= +github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8/go.mod h1:HUYIGzjTL3rfEspMxjDjgmT5uz5wzYJKVo23qUhYTos= +go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= +go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q= +go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ= +go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I= +go.opentelemetry.io/otel/metric v1.37.0 h1:mvwbQS5m0tbmqML4NqK+e3aDiO02vsf/WgbsdpcPoZE= +go.opentelemetry.io/otel/metric v1.37.0/go.mod h1:04wGrZurHYKOc+RKeye86GwKiTb9FKm1WHtO+4EVr2E= +go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= +go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= golang.org/x/oauth2 v0.30.0 h1:dnDm7JmhM45NNpd8FDDeLhK6FwqbOf4MLCM9zb1BOHI= golang.org/x/oauth2 v0.30.0/go.mod h1:B++QgG3ZKulg6sRPGD/mqlHQs5rB3Ml9erfeDY7xKlU= +golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= +golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= +golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= +golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/adapters/store/migrations/001_initial.down.sql b/internal/adapters/store/migrations/001_initial.down.sql new file mode 100644 index 0000000..82a9138 --- /dev/null +++ b/internal/adapters/store/migrations/001_initial.down.sql @@ -0,0 +1,5 @@ +DROP TABLE IF EXISTS sink_deliveries; +DROP TABLE IF EXISTS summaries; +DROP TABLE IF EXISTS transcripts; +DROP TABLE IF EXISTS videos; +DROP TABLE IF EXISTS users; diff --git a/internal/adapters/store/migrations/001_initial.up.sql b/internal/adapters/store/migrations/001_initial.up.sql new file mode 100644 index 0000000..48992c0 --- /dev/null +++ b/internal/adapters/store/migrations/001_initial.up.sql @@ -0,0 +1,73 @@ +-- Tapir initial schema (Stage 0). Per-user isolation: every user-owned table +-- carries user_id even though Stage 0 has a single user (docs/data-model.md). +-- Scoped to Stage 0 dedup + delivery; Future B/C concerns are out of scope. + +CREATE TABLE users ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + display_name TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- One row per (user, video) — per-user isolation, not a global deduped table +-- (DECISIONS.md "Rejected alternatives"). subscription_id has no FK at Stage 0: +-- the subscriptions table is not part of the store-sink slice. +CREATE TABLE videos ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + subscription_id UUID, + provider TEXT NOT NULL, + provider_video_id TEXT NOT NULL, + title TEXT, + duration_s INTEGER, + published_at TIMESTAMPTZ, + url TEXT, + seen_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT videos_user_provider_video_unique UNIQUE (user_id, provider, provider_video_id) +); + +CREATE INDEX idx_videos_user_id ON videos(user_id); + +-- At most one transcript per video. source = 'none' records "checked, no usable +-- transcript" so the watcher does not reprocess (ADR-007); content is NULL then. +CREATE TABLE transcripts ( + video_id UUID PRIMARY KEY REFERENCES videos(id) ON DELETE CASCADE, + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + source TEXT NOT NULL, + language TEXT, + content TEXT, + resolved_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX idx_transcripts_user_id ON transcripts(user_id); + +-- At most one summary per video (UNIQUE on (user_id, video_id)). video_id is not +-- FK-constrained to videos at Stage 0: the store sink receives only a Summary +-- (ports.Sink.Deliver), so the durable dedup key (user_id, video_id) stands on +-- its own; video-row persistence is the engine/source's concern, deferred. +CREATE TABLE summaries ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + video_id UUID NOT NULL, + summary TEXT NOT NULL, + highlights JSONB NOT NULL DEFAULT '[]'::jsonb, + takeaways JSONB NOT NULL DEFAULT '[]'::jsonb, + ai_provider TEXT, + ai_model TEXT, + fallback_used BOOLEAN NOT NULL DEFAULT FALSE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT summaries_user_video_unique UNIQUE (user_id, video_id) +); + +CREATE INDEX idx_summaries_user_id ON summaries(user_id); + +-- One row per (summary, sink) attempt. "Also sent to brain" lives here as a +-- delivery row with sink = 'brain'; no brain-specific tables (data-model.md). +CREATE TABLE sink_deliveries ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + summary_id UUID NOT NULL REFERENCES summaries(id) ON DELETE CASCADE, + sink TEXT NOT NULL, + status TEXT NOT NULL, + detail TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT sink_deliveries_summary_sink_unique UNIQUE (summary_id, sink) +); diff --git a/internal/adapters/store/store.go b/internal/adapters/store/store.go new file mode 100644 index 0000000..16737d4 --- /dev/null +++ b/internal/adapters/store/store.go @@ -0,0 +1,201 @@ +// Package store is the Postgres-backed implementation of ports.Sink: the user's +// own durable store of summaries. It is the primary sink (ADR-003) and the source +// of the engine's durable, cross-restart dedup (the in-engine map is process- +// lifetime only). Connection is pgx/v5 + pgxpool with the DSN from the caller; +// schema is applied via golang-migrate from embedded migrations. +// +// Per-user isolation (docs/data-model.md) is a Stage-0 promise: every row carries +// user_id and every read is scoped by it, even though the demo has one user. +package store + +import ( + "context" + "database/sql" + "embed" + "encoding/json" + "errors" + "fmt" + + "github.com/golang-migrate/migrate/v4" + migratepgx "github.com/golang-migrate/migrate/v4/database/pgx/v5" + "github.com/golang-migrate/migrate/v4/source/iofs" + "github.com/jackc/pgx/v5/pgxpool" + + _ "github.com/jackc/pgx/v5/stdlib" // register the "pgx" database/sql driver for migrate + + "gitea.d-ma.be/mathias/tapir/internal/domain" +) + +//go:embed migrations/*.sql +var migrationsFS embed.FS + +// Store persists summaries to Postgres and answers durable dedup queries. +type Store struct { + pool *pgxpool.Pool +} + +// New connects a pool to dsn, applies all pending migrations, and verifies the +// connection. The caller owns the lifetime: call Close when done. +func New(ctx context.Context, dsn string) (*Store, error) { + if dsn == "" { + return nil, errors.New("store: empty DSN") + } + if err := Migrate(dsn); err != nil { + return nil, err + } + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + return nil, fmt.Errorf("store: connect pool: %w", err) + } + if err := pool.Ping(ctx); err != nil { + pool.Close() + return nil, fmt.Errorf("store: ping: %w", err) + } + return &Store{pool: pool}, nil +} + +// Migrate applies all pending up-migrations against dsn. It opens its own +// short-lived connection (golang-migrate uses database/sql) and closes it before +// returning, so it can run before the pool is created or be invoked standalone. +func Migrate(dsn string) error { + db, err := sql.Open("pgx", dsn) + if err != nil { + return fmt.Errorf("store: open migrate db: %w", err) + } + defer func() { _ = db.Close() }() + + drv, err := migratepgx.WithInstance(db, &migratepgx.Config{}) + if err != nil { + return fmt.Errorf("store: migrate driver: %w", err) + } + src, err := iofs.New(migrationsFS, "migrations") + if err != nil { + return fmt.Errorf("store: migrate source: %w", err) + } + m, err := migrate.NewWithInstance("iofs", src, "pgx", drv) + if err != nil { + return fmt.Errorf("store: migrator: %w", err) + } + if err := m.Up(); err != nil && !errors.Is(err, migrate.ErrNoChange) { + return fmt.Errorf("store: migrate up: %w", err) + } + return nil +} + +// Close releases the connection pool. +func (s *Store) Close() { + s.pool.Close() +} + +// Name identifies this sink in delivery records. +func (s *Store) Name() string { return "store" } + +// Deliver upserts the summary idempotently on (user_id, video_id) and records the +// store delivery. Re-delivering the same summary updates in place — it never +// errors or duplicates. The whole write is one transaction so a summary and its +// delivery row stay consistent. +func (s *Store) Deliver(ctx context.Context, sum domain.Summary) error { + highlights, err := marshalList(sum.Highlights) + if err != nil { + return fmt.Errorf("store: marshal highlights: %w", err) + } + takeaways, err := marshalList(sum.Takeaways) + if err != nil { + return fmt.Errorf("store: marshal takeaways: %w", err) + } + + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("store: begin: %w", err) + } + defer tx.Rollback(ctx) //nolint:errcheck // no-op after Commit + + // Ensure the owning user exists (FK target). The store sink receives only a + // Summary, so a minimal user row is enough at Stage 0. + if _, err := tx.Exec(ctx, + `INSERT INTO users (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, + sum.UserID); err != nil { + return fmt.Errorf("store: upsert user: %w", err) + } + + var summaryID string + if err := tx.QueryRow(ctx, + `INSERT INTO summaries + (user_id, video_id, summary, highlights, takeaways, ai_provider, ai_model, fallback_used) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + ON CONFLICT (user_id, video_id) DO UPDATE SET + summary = EXCLUDED.summary, + highlights = EXCLUDED.highlights, + takeaways = EXCLUDED.takeaways, + ai_provider = EXCLUDED.ai_provider, + ai_model = EXCLUDED.ai_model, + fallback_used = EXCLUDED.fallback_used + RETURNING id`, + sum.UserID, sum.VideoID, sum.Summary, highlights, takeaways, + sum.AIProvider, sum.AIModel, sum.FallbackUsed, + ).Scan(&summaryID); err != nil { + return fmt.Errorf("store: upsert summary: %w", err) + } + + if _, err := tx.Exec(ctx, + `INSERT INTO sink_deliveries (summary_id, sink, status) + VALUES ($1, 'store', 'delivered') + ON CONFLICT (summary_id, sink) DO UPDATE SET + status = 'delivered', + detail = NULL, + updated_at = NOW()`, + summaryID); err != nil { + return fmt.Errorf("store: record delivery: %w", err) + } + + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("store: commit: %w", err) + } + return nil +} + +// HasSummary reports whether a summary already exists for (userID, videoID). +// This is the per-video durable dedup check. +func (s *Store) HasSummary(ctx context.Context, userID, videoID string) (bool, error) { + var exists bool + if err := s.pool.QueryRow(ctx, + `SELECT EXISTS(SELECT 1 FROM summaries WHERE user_id = $1 AND video_id = $2)`, + userID, videoID).Scan(&exists); err != nil { + return false, fmt.Errorf("store: has summary: %w", err) + } + return exists, nil +} + +// SeenVideoIDs returns the set of video IDs that already have a summary for the +// user. The watcher uses it to skip re-summarizing across restarts. Scoped by +// user_id, so one user never sees another's videos. +func (s *Store) SeenVideoIDs(ctx context.Context, userID string) (map[string]bool, error) { + rows, err := s.pool.Query(ctx, + `SELECT video_id FROM summaries WHERE user_id = $1`, userID) + if err != nil { + return nil, fmt.Errorf("store: seen video ids: %w", err) + } + defer rows.Close() + + seen := make(map[string]bool) + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("store: scan video id: %w", err) + } + seen[id] = true + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("store: iterate video ids: %w", err) + } + return seen, nil +} + +// marshalList renders a string slice as a JSON array, normalising nil to "[]" so +// the jsonb columns never hold SQL/JSON null. +func marshalList(xs []string) ([]byte, error) { + if xs == nil { + xs = []string{} + } + return json.Marshal(xs) +} diff --git a/internal/adapters/store/store_test.go b/internal/adapters/store/store_test.go new file mode 100644 index 0000000..cac0e81 --- /dev/null +++ b/internal/adapters/store/store_test.go @@ -0,0 +1,173 @@ +package store_test + +import ( + "context" + "fmt" + "os" + "testing" + + embeddedpostgres "github.com/fergusstrange/embedded-postgres" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" + + "gitea.d-ma.be/mathias/tapir/internal/adapters/store" + "gitea.d-ma.be/mathias/tapir/internal/domain" + "gitea.d-ma.be/mathias/tapir/internal/ports" +) + +// Static check: Store satisfies the Sink port. +var _ ports.Sink = (*store.Store)(nil) + +// dsn points at the in-process Postgres started in TestMain. Tests run against +// real SQL (constraints, ON CONFLICT, jsonb) — not a mock — without docker or +// live-cluster credentials (embedded-postgres downloads its own PG binary). +var dsn string + +func TestMain(m *testing.M) { + const port = 54329 + dsn = fmt.Sprintf("postgres://postgres:postgres@localhost:%d/postgres?sslmode=disable", port) + + pg := embeddedpostgres.NewDatabase( + embeddedpostgres.DefaultConfig().Port(port), + ) + if err := pg.Start(); err != nil { + fmt.Fprintf(os.Stderr, "embedded-postgres start: %v\n", err) + os.Exit(1) + } + + code := m.Run() + + if err := pg.Stop(); err != nil { + fmt.Fprintf(os.Stderr, "embedded-postgres stop: %v\n", err) + } + os.Exit(code) +} + +// uuids — fixed so tests are deterministic. user_id/video_id are UUID columns. +const ( + userA = "11111111-1111-1111-1111-111111111111" + userB = "22222222-2222-2222-2222-222222222222" + videoX = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" + videoY = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" +) + +func newStore(t *testing.T) *store.Store { + t.Helper() + s, err := store.New(context.Background(), dsn) + require.NoError(t, err, "migrations must apply clean and the pool must connect") + t.Cleanup(s.Close) + return s +} + +// rawPool is a direct connection for test introspection (counts, truncation), +// kept out of the production Store API. Closed via t.Cleanup. +func rawPool(t *testing.T) *pgxpool.Pool { + t.Helper() + p, err := pgxpool.New(context.Background(), dsn) + require.NoError(t, err) + t.Cleanup(p.Close) + return p +} + +// resetDB truncates between tests so each starts from a known state. Schema is +// shared across the run (migrations are idempotent via New). +func resetDB(t *testing.T, p *pgxpool.Pool) { + t.Helper() + _, err := p.Exec(context.Background(), + `TRUNCATE sink_deliveries, summaries, transcripts, videos, users CASCADE`) + require.NoError(t, err) +} + +func summary(userID, videoID, text string) domain.Summary { + return domain.Summary{ + UserID: userID, + VideoID: videoID, + Summary: text, + Highlights: []string{"h1", "h2"}, + Takeaways: []string{"t1"}, + AIProvider: "local", + AIModel: "qwen", + } +} + +func TestName(t *testing.T) { + require.Equal(t, "store", newStore(t).Name()) +} + +func TestDeliverInsertsSummary(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "first"))) + + ok, err := s.HasSummary(ctx, userA, videoX) + require.NoError(t, err) + require.True(t, ok) +} + +func TestDeliverIsIdempotentOnUserVideo(t *testing.T) { + ctx := context.Background() + s := newStore(t) + p := rawPool(t) + resetDB(t, p) + + require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "first"))) + // Re-deliver the same (user, video): must update in place, not duplicate or error. + require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "second"))) + + var count int + require.NoError(t, p.QueryRow(ctx, + `SELECT count(*) FROM summaries WHERE user_id = $1 AND video_id = $2`, + userA, videoX).Scan(&count)) + require.Equal(t, 1, count, "second delivery must update, not duplicate") + + var text string + require.NoError(t, p.QueryRow(ctx, + `SELECT summary FROM summaries WHERE user_id = $1 AND video_id = $2`, + userA, videoX).Scan(&text)) + require.Equal(t, "second", text, "second delivery must overwrite the summary text") + + // Exactly one store-delivery row for the summary (ON CONFLICT update). + var deliveries int + require.NoError(t, p.QueryRow(ctx, + `SELECT count(*) FROM sink_deliveries d + JOIN summaries m ON m.id = d.summary_id + WHERE m.user_id = $1 AND m.video_id = $2 AND d.sink = 'store'`, + userA, videoX).Scan(&deliveries)) + require.Equal(t, 1, deliveries) +} + +func TestSeenVideoIDsReturnsUsersSet(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "x"))) + require.NoError(t, s.Deliver(ctx, summary(userA, videoY, "y"))) + + seen, err := s.SeenVideoIDs(ctx, userA) + require.NoError(t, err) + require.Equal(t, map[string]bool{videoX: true, videoY: true}, seen) +} + +func TestPerUserIsolation(t *testing.T) { + ctx := context.Background() + s := newStore(t) + resetDB(t, rawPool(t)) + + require.NoError(t, s.Deliver(ctx, summary(userA, videoX, "a-owns-this"))) + + // User B must not see user A's video, by either dedup read. + seenB, err := s.SeenVideoIDs(ctx, userB) + require.NoError(t, err) + require.Empty(t, seenB, "user B must not see user A's videos") + + hasB, err := s.HasSummary(ctx, userB, videoX) + require.NoError(t, err) + require.False(t, hasB, "the same video_id under another user must be invisible") + + hasA, err := s.HasSummary(ctx, userA, videoX) + require.NoError(t, err) + require.True(t, hasA) +}