diff --git a/internal/store/README.md b/internal/store/README.md index da99a43..0476500 100644 --- a/internal/store/README.md +++ b/internal/store/README.md @@ -36,7 +36,7 @@ internal/store/common.go common for migrate, store and direct access | `opportunity_marks` | `uuid` | `uuid` references `opportunities`. The user's own marks (`hidden`, `skipped`, `interested`, `bid_submitted`, `won`, `lost`), each `BOOLEAN NOT NULL DEFAULT 0`. A `CHECK` constraint stops `won` and `lost` both being set. | | `opportunity_releases` | (`uuid`, `release_id`) | Write-once. `uuid` references `opportunities`. `data` is TEXT, so `json_extract` works on it. | | `opportunity_release_tags` | (`uuid`, `release_id`, `tag`) | Write-once. (`uuid`, `release_id`) references `opportunity_releases`. | -| `classification_codes` | `id` | Shared between opportunities. A row is reused only when every content column matches. The table grows over time to include codes not already seen in fetched opportunities. | +| `classification_codes` | `id` | Shared between opportunities. A CPV code has one row (unique on `code` where `is_cpv`), keeping the longest description it has been given: EU Supply cuts descriptions to 64 characters. A code of any other scheme gets a row per distinct content. The table grows over time to include codes not already seen in fetched opportunities. | | `opportunity_classifications` | (`opportunity_uuid`, `classification_id`) | Mapping table; replaced as a set when the opportunity is updated. | | `opportunities_fts` | rowid = `opportunities.id` | FTS5 index on `opportunities.title`, kept in sync by triggers. | | `opportunity_change_batches` | `id`; `hash` (unique) | Write-once. One row per save that adds opportunities or gives them new releases, and per change of marks. `hash` names the batch to clients (random, so unguessable); `created_at` orders the batches. | @@ -44,7 +44,7 @@ internal/store/common.go common for migrate, store and direct access Foreign keys are enforced (`_foreign_keys=on` in the connection string) and cascade on delete from `opportunities`. The connection also enables WAL mode, a 5s busy timeout, and `_txlock=immediate`, so a save takes the write lock up front rather than failing halfway through. -**When adding a classification scheme flag** (e.g. `is_unspsc`), the migration adding the column must drop and recreate the `classification_codes_content` unique index with the new column included. Otherwise rows differing only by that flag would be merged. +**When adding a classification scheme flag** (e.g. `is_unspsc`), the migration adding the column must drop and recreate the `classification_codes_content` unique index with the new column included. Otherwise rows differing only by that flag would be merged. A scheme whose codes have one official description, like CPV, can instead have a partial unique index on `code` (as `classification_codes_cpv` does), with its own `ON CONFLICT` clause in `upsertClassificationCodeReturningID`. **internal/types**: The structs contain `db` tags to allow sqlx to map table columns to struct members. Any struct members not mapped to a column (including child slices) are referenced by `db:"-"`. diff --git a/internal/store/migrate_test.go b/internal/store/migrate_test.go index c7cf115..141f8bf 100644 --- a/internal/store/migrate_test.go +++ b/internal/store/migrate_test.go @@ -327,3 +327,78 @@ func TestFailedMigrationIsRolledBack(t *testing.T) { t.Errorf("failed migration left %d objects behind", indexes) } } + +// Migration 10 merges each CPV code's rows into the one with the longest +// description, moving the links to it (one link if an opportunity had both), +// and leaves other schemes' rows alone. +func TestMigration10MergesCPVDuplicates(t *testing.T) { + ctx := context.Background() + m, db := newMigrator(t) + if _, err := m.UpTo(ctx, 9); err != nil { + t.Fatal(err) + } + + const full = "Petroleum products, fuel, electricity and other sources of energy" + setup := []string{ + `INSERT INTO opportunities (uuid, uuid_type, title, status, latest_release_date) + VALUES ('a', 'ocid', 'A', 'active', '2026-10-01'), + ('b', 'ocid', 'B', 'active', '2026-10-01')`, + `INSERT INTO classification_codes (id, is_cpv, code, classification) VALUES + (1, 1, '09000000', '` + full + `'), + (2, 1, '09000000', '` + full[:64] + `'), + (3, 0, '123', 'Scheme X'), + (4, 0, '123', 'Scheme X, amended')`, + // a links to both CPV rows, b only to the cut one. + `INSERT INTO opportunity_classifications VALUES + ('a', 1), ('a', 2), ('b', 2), ('a', 3), ('b', 4)`, + } + for _, statement := range setup { + if _, err := db.ExecContext(ctx, statement); err != nil { + t.Fatal(err) + } + } + + if _, err := m.UpTo(ctx, 10); err != nil { + t.Fatal(err) + } + + var codes []string + rows, err := db.QueryContext(ctx, + `SELECT id || ':' || classification FROM classification_codes ORDER BY id`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var code string + rows.Scan(&code) + codes = append(codes, code) + } + rows.Close() + want := "1:" + full + "|3:Scheme X|4:Scheme X, amended" + if got := strings.Join(codes, "|"); got != want { + t.Errorf("codes = %s; want %s", got, want) + } + + var links []string + rows, err = db.QueryContext(ctx, `SELECT opportunity_uuid || classification_id + FROM opportunity_classifications ORDER BY 1`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var link string + rows.Scan(&link) + links = append(links, link) + } + rows.Close() + if got := strings.Join(links, ","); got != "a1,a3,b1,b4" { + t.Errorf("links = %s; want a1,a3,b1,b4", got) + } + + // A second row for the code is now refused. + _, err = db.ExecContext(ctx, `INSERT INTO classification_codes + (is_cpv, code, classification) VALUES (1, '09000000', 'Other')`) + if err == nil { + t.Error("a second row for a CPV code was accepted") + } +} diff --git a/internal/store/migrations/00010_unique_cpv_codes.sql b/internal/store/migrations/00010_unique_cpv_codes.sql new file mode 100644 index 0000000..029529c --- /dev/null +++ b/internal/store/migrations/00010_unique_cpv_codes.sql @@ -0,0 +1,60 @@ +-- +goose Up +-- +goose StatementBegin +-- A CPV code has one official description, so it is now one row whatever +-- description it arrives with: EU Supply cuts descriptions to 64 characters, +-- which made a second row for each code it sent. The duplicates are merged +-- into the row with the longest description, and the upsert keeps the +-- longest from now on. Other schemes still get a row per distinct content. + +-- The row each CPV row merges into. +CREATE TEMP TABLE "cpv_merge" AS +SELECT + "c"."id" AS "id", + ( + SELECT "k"."id" FROM "classification_codes" "k" + WHERE "k"."is_cpv" AND "k"."code" = "c"."code" + ORDER BY length("k"."classification") DESC, "k"."id" + LIMIT 1 + ) AS "keep_id" +FROM "classification_codes" "c" +WHERE "c"."is_cpv"; + +-- An opportunity linked to both rows keeps only its link to the kept one. +DELETE FROM "opportunity_classifications" +WHERE "classification_id" IN ( + SELECT "id" FROM "cpv_merge" WHERE "id" <> "keep_id" +) +AND EXISTS ( + SELECT 1 + FROM "cpv_merge" "m" + JOIN "opportunity_classifications" "kept" + ON "kept"."classification_id" = "m"."keep_id" + WHERE "m"."id" = "opportunity_classifications"."classification_id" + AND "kept"."opportunity_uuid" + = "opportunity_classifications"."opportunity_uuid" +); + +UPDATE "opportunity_classifications" +SET "classification_id" = ( + SELECT "keep_id" FROM "cpv_merge" + WHERE "id" = "opportunity_classifications"."classification_id" +) +WHERE "classification_id" IN ( + SELECT "id" FROM "cpv_merge" WHERE "id" <> "keep_id" +); + +DELETE FROM "classification_codes" +WHERE "id" IN (SELECT "id" FROM "cpv_merge" WHERE "id" <> "keep_id"); + +DROP TABLE "cpv_merge"; + +CREATE UNIQUE INDEX IF NOT EXISTS "classification_codes_cpv" + ON "classification_codes" ("code") WHERE "is_cpv"; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +-- The merged duplicates aren't restored: they only differed by a cut-off +-- description. +DROP INDEX IF EXISTS "classification_codes_cpv"; +-- +goose StatementEnd diff --git a/internal/store/queries.go b/internal/store/queries.go index d871765..f9cb021 100644 --- a/internal/store/queries.go +++ b/internal/store/queries.go @@ -197,15 +197,24 @@ ORDER BY release_id, tag` // ---------- classification_codes ---------- -// upsertClassificationCodeReturningID returns the id of the row matching -// every content column, inserting it first if needed. Named parameters +// upsertClassificationCodeReturningID returns the id of the row for a +// classification, inserting it first if needed. Named parameters // (types.ClassificationCode). // -// The no-op DO UPDATE (rather than DO NOTHING) makes RETURNING yield the -// existing row's id on conflict. +// A CPV code has one row whatever its description (migration 00010): the +// row keeps the longest description it has been given, as EU Supply cuts +// them to 64 characters. Any other scheme's row matches every content +// column. The no-op DO UPDATE (rather than DO NOTHING) makes RETURNING +// yield the existing row's id on conflict. const upsertClassificationCodeReturningID = ` INSERT INTO classification_codes (is_cpv, code, classification) VALUES (:is_cpv, :code, :classification) +ON CONFLICT (code) WHERE is_cpv + DO UPDATE SET classification = CASE + WHEN length(excluded.classification) > length(classification) + THEN excluded.classification + ELSE classification + END ON CONFLICT (is_cpv, code, classification) DO UPDATE SET is_cpv = excluded.is_cpv RETURNING id` diff --git a/internal/store/store_test.go b/internal/store/store_test.go index 6a28d74..066c794 100644 --- a/internal/store/store_test.go +++ b/internal/store/store_test.go @@ -3,6 +3,7 @@ package store_test import ( "context" "errors" + "fmt" "path/filepath" "slices" "strings" @@ -273,8 +274,9 @@ func TestNewerReleaseUpdatesOpportunityAndSearchIndex(t *testing.T) { got.Classifications[0].Classification != "IT services (amended)" { t.Fatalf("classifications not replaced: %+v", got.Classifications) } - if got.Classifications[0].ID == oldIT { - t.Errorf("changed description should create a new classification row") + // A CPV code is one row: the longer description replaces the shorter. + if got.Classifications[0].ID != oldIT { + t.Errorf("a CPV code's new description should keep its row") } for title, want := range map[string]int{"cloud": 0, "redesign": 1} { @@ -295,6 +297,61 @@ func TestNewerReleaseUpdatesOpportunityAndSearchIndex(t *testing.T) { } } +// A CPV code keeps one row and its longest description, whichever order +// the descriptions arrive in (EU Supply cuts them to 64 characters); a code +// of another scheme still gets a row per description. +func TestCPVCodeKeepsLongestDescription(t *testing.T) { + const full = "Petroleum products, fuel, electricity and other sources of energy" + cut := full[:64] + for name, order := range map[string][]string{ + "full first": {full, cut}, + "cut first": {cut, full}, + } { + t.Run(name, func(t *testing.T) { + s := openStore(t) + var ids []int64 + for i, description := range order { + o := opportunity(fmt.Sprintf("ocds-%d", i), "Fuel", 1, "r1") + o.Classifications = []types.ClassificationCode{ + { + IsCPV: true, + Code: "09000000", + Classification: description, + }, + } + mustSave(t, s, o) + ids = append(ids, o.Classifications[0].ID) + } + if ids[0] != ids[1] { + t.Errorf("ids = %v; want one row for the code", ids) + } + for _, uuid := range []string{"ocds-0", "ocds-1"} { + got := mustGet(t, s, uuid).Classifications[0].Classification + if got != full { + t.Errorf("%s: description = %q; want the full one", uuid, got) + } + } + }) + } + + s := openStore(t) + var ids []int64 + for i, description := range []string{"Scheme X", "Scheme X, amended"} { + o := opportunity(fmt.Sprintf("other-%d", i), "Other", 1, "r1") + o.Classifications = []types.ClassificationCode{ + { + Code: "123", + Classification: description, + }, + } + mustSave(t, s, o) + ids = append(ids, o.Classifications[0].ID) + } + if ids[0] == ids[1] { + t.Errorf("a non-CPV code's descriptions share a row: %v", ids) + } +} + func TestClassificationRowsAreShared(t *testing.T) { s := openStore(t) a := opportunity("ocds-a", "A", 1, "r1")