From ed3c601a9dc72bbef3cb4fce02cfc139691d68f8 Mon Sep 17 00:00:00 2001 From: riddim-developer-bot Date: Mon, 25 May 2026 15:02:49 -0400 Subject: [PATCH] Build Hansard search index Lambda --- .github/workflows/backend-pr.yml | 1 + .github/workflows/backend-production.yml | 10 +- .github/workflows/backend-staging.yml | 13 +- .github/workflows/backend-tests.yml | 2 +- .github/workflows/hansard-search-reindex.yml | 126 +++++++ backend/go.work | 1 + backend/hansard-search-index/.gitignore | 4 + backend/hansard-search-index/README.md | 141 ++++++++ backend/hansard-search-index/go.mod | 46 +++ backend/hansard-search-index/go.sum | 113 ++++++ .../internal/adapter/ourcommons/log.go | 67 ++++ .../internal/adapter/ourcommons/parser.go | 319 ++++++++++++++++ .../adapter/ourcommons/parser_test.go | 69 ++++ .../internal/adapter/ourcommons/source.go | 234 ++++++++++++ .../adapter/ourcommons/source_test.go | 114 ++++++ .../internal/adapter/s3/s3.go | 114 ++++++ .../internal/adapter/s3/s3_test.go | 76 ++++ .../internal/adapter/sqlitefts5/builder.go | 342 ++++++++++++++++++ .../adapter/sqlitefts5/builder_test.go | 130 +++++++ .../internal/domain/domain.go | 51 +++ .../internal/usecase/usecase.go | 172 +++++++++ .../internal/usecase/usecase_test.go | 150 ++++++++ backend/hansard-search-index/main.go | 126 +++++++ backend/hansard-search-index/main_test.go | 39 ++ .../testdata/hansard_451_001_slice.xml | 37 ++ backend/manifest/deployment-services.json | 8 + docs/architecture/use-case-catalog.md | 30 ++ infra/terraform/core/iam.tf | 27 +- infra/terraform/core/variables.tf | 12 + infra/terraform/production/main.tf | 15 +- infra/terraform/staging/main.tf | 9 + scripts/ci/backend_staging_smoke.py | 92 +++++ 32 files changed, 2685 insertions(+), 5 deletions(-) create mode 100644 .github/workflows/hansard-search-reindex.yml create mode 100644 backend/hansard-search-index/.gitignore create mode 100644 backend/hansard-search-index/README.md create mode 100644 backend/hansard-search-index/go.mod create mode 100644 backend/hansard-search-index/go.sum create mode 100644 backend/hansard-search-index/internal/adapter/ourcommons/log.go create mode 100644 backend/hansard-search-index/internal/adapter/ourcommons/parser.go create mode 100644 backend/hansard-search-index/internal/adapter/ourcommons/parser_test.go create mode 100644 backend/hansard-search-index/internal/adapter/ourcommons/source.go create mode 100644 backend/hansard-search-index/internal/adapter/ourcommons/source_test.go create mode 100644 backend/hansard-search-index/internal/adapter/s3/s3.go create mode 100644 backend/hansard-search-index/internal/adapter/s3/s3_test.go create mode 100644 backend/hansard-search-index/internal/adapter/sqlitefts5/builder.go create mode 100644 backend/hansard-search-index/internal/adapter/sqlitefts5/builder_test.go create mode 100644 backend/hansard-search-index/internal/domain/domain.go create mode 100644 backend/hansard-search-index/internal/usecase/usecase.go create mode 100644 backend/hansard-search-index/internal/usecase/usecase_test.go create mode 100644 backend/hansard-search-index/main.go create mode 100644 backend/hansard-search-index/main_test.go create mode 100644 backend/hansard-search-index/testdata/hansard_451_001_slice.xml diff --git a/.github/workflows/backend-pr.yml b/.github/workflows/backend-pr.yml index 7900076c..35aacf61 100644 --- a/.github/workflows/backend-pr.yml +++ b/.github/workflows/backend-pr.yml @@ -47,6 +47,7 @@ jobs: - calendar - config - hansard-subjects-index + - hansard-search-index - manifest - openapi - members-publisher diff --git a/.github/workflows/backend-production.yml b/.github/workflows/backend-production.yml index 192d98d7..adac4ea2 100644 --- a/.github/workflows/backend-production.yml +++ b/.github/workflows/backend-production.yml @@ -209,10 +209,14 @@ jobs: - name: Sync production artifact environment if: ${{ (inputs.service == 'all' || inputs.service == matrix.service) && contains(fromJson(needs.prepare.outputs.artifact_services), matrix.service) }} env: + SERVICE: ${{ matrix.service }} CANONICAL_FUNCTION_NAME: epac-${{ matrix.service }}-production LEGACY_FUNCTION_NAME: ${{ matrix.service }} ARTIFACT_BUCKET: ${{ vars.EPAC_ARTIFACT_BUCKET_PRODUCTION || vars.EPAC_ARTIFACT_BUCKET || vars.ARTIFACTS_BUCKET }} ARTIFACT_PREFIX: ${{ vars.EPAC_ARTIFACT_PREFIX_PRODUCTION || vars.EPAC_ARTIFACT_PREFIX }} + HANSARD_SEARCH_PREFIX: ${{ vars.EPAC_HANSARD_SEARCH_PREFIX || 'hansard-search/v1' }} + PARLIAMENT_NUMBER: ${{ vars.EPAC_HANSARD_PARLIAMENT_NUMBER || '45' }} + SESSION_NUMBER: ${{ vars.EPAC_HANSARD_SESSION_NUMBER || '1' }} SENTRY_DSN: ${{ secrets.SENTRY_DSN_PRODUCTION }} run: | set -euo pipefail @@ -232,10 +236,14 @@ jobs: --output json) UPDATED_ENV=$(jq -cn \ --argjson current "${CURRENT_ENV:-null}" \ + --arg service "$SERVICE" \ --arg artifact_bucket "$ARTIFACT_BUCKET" \ --arg artifact_prefix "$ARTIFACT_PREFIX" \ + --arg hansard_search_prefix "$HANSARD_SEARCH_PREFIX" \ + --arg parliament_number "$PARLIAMENT_NUMBER" \ + --arg session_number "$SESSION_NUMBER" \ --arg sentry_dsn "$SENTRY_DSN" \ - '{Variables: ((($current // {}) | del(.DATABASE_URL)) + {EPAC_ARTIFACT_BUCKET: $artifact_bucket} + (if $artifact_prefix != "" then {EPAC_ARTIFACT_PREFIX: $artifact_prefix} else {} end) + (if $sentry_dsn != "" then {SENTRY_DSN: $sentry_dsn} else {} end))}') + '{Variables: ((($current // {}) | del(.DATABASE_URL)) + {EPAC_ARTIFACT_BUCKET: $artifact_bucket} + (if $artifact_prefix != "" then {EPAC_ARTIFACT_PREFIX: $artifact_prefix} else {} end) + (if $service == "hansard-search-index" then {EPAC_HANSARD_SEARCH_PREFIX: $hansard_search_prefix, PARLIAMENT_NUMBER: $parliament_number, SESSION_NUMBER: $session_number} else {} end) + (if $sentry_dsn != "" then {SENTRY_DSN: $sentry_dsn} else {} end))}') aws lambda update-function-configuration \ --function-name "$FUNCTION_NAME" \ diff --git a/.github/workflows/backend-staging.yml b/.github/workflows/backend-staging.yml index 7ddfa34b..2470f67d 100644 --- a/.github/workflows/backend-staging.yml +++ b/.github/workflows/backend-staging.yml @@ -209,9 +209,13 @@ jobs: - name: Sync staging artifact environment if: ${{ contains(fromJson(needs.prepare.outputs.artifact_services), matrix.service) }} env: + SERVICE: ${{ matrix.service }} FUNCTION_NAME: epac-${{ matrix.service }}-staging ARTIFACT_BUCKET: ${{ vars.EPAC_ARTIFACT_BUCKET_STAGING || vars.EPAC_ARTIFACT_BUCKET || vars.ARTIFACTS_BUCKET }} ARTIFACT_PREFIX: ${{ vars.EPAC_ARTIFACT_PREFIX_STAGING || vars.EPAC_ARTIFACT_PREFIX }} + HANSARD_SEARCH_PREFIX: ${{ vars.EPAC_HANSARD_SEARCH_PREFIX || 'hansard-search/v1' }} + PARLIAMENT_NUMBER: ${{ vars.EPAC_HANSARD_PARLIAMENT_NUMBER || '45' }} + SESSION_NUMBER: ${{ vars.EPAC_HANSARD_SESSION_NUMBER || '1' }} SENTRY_DSN: ${{ secrets.SENTRY_DSN_STAGING }} run: | set -euo pipefail @@ -227,10 +231,14 @@ jobs: --output json) UPDATED_ENV=$(jq -cn \ --argjson current "${CURRENT_ENV:-null}" \ + --arg service "$SERVICE" \ --arg artifact_bucket "$ARTIFACT_BUCKET" \ --arg artifact_prefix "$ARTIFACT_PREFIX" \ + --arg hansard_search_prefix "$HANSARD_SEARCH_PREFIX" \ + --arg parliament_number "$PARLIAMENT_NUMBER" \ + --arg session_number "$SESSION_NUMBER" \ --arg sentry_dsn "$SENTRY_DSN" \ - '{Variables: ((($current // {}) | del(.DATABASE_URL)) + {EPAC_ARTIFACT_BUCKET: $artifact_bucket} + (if $artifact_prefix != "" then {EPAC_ARTIFACT_PREFIX: $artifact_prefix} else {} end) + (if $sentry_dsn != "" then {SENTRY_DSN: $sentry_dsn} else {} end))}') + '{Variables: ((($current // {}) | del(.DATABASE_URL)) + {EPAC_ARTIFACT_BUCKET: $artifact_bucket} + (if $artifact_prefix != "" then {EPAC_ARTIFACT_PREFIX: $artifact_prefix} else {} end) + (if $service == "hansard-search-index" then {EPAC_HANSARD_SEARCH_PREFIX: $hansard_search_prefix, PARLIAMENT_NUMBER: $parliament_number, SESSION_NUMBER: $session_number} else {} end) + (if $sentry_dsn != "" then {SENTRY_DSN: $sentry_dsn} else {} end))}') aws lambda update-function-configuration \ --function-name "$FUNCTION_NAME" \ @@ -265,4 +273,7 @@ jobs: run: ./scripts/ci/ensure-staging-api-routes.sh - name: Run staging backend smoke tests + env: + EPAC_ARTIFACT_BUCKET: ${{ vars.EPAC_ARTIFACT_BUCKET_STAGING || vars.EPAC_ARTIFACT_BUCKET || vars.ARTIFACTS_BUCKET }} + EPAC_HANSARD_SEARCH_PREFIX: ${{ vars.EPAC_HANSARD_SEARCH_PREFIX || 'hansard-search/v1' }} run: python3 scripts/ci/backend_staging_smoke.py --base-url "$STAGING_API_BASE_URL" diff --git a/.github/workflows/backend-tests.yml b/.github/workflows/backend-tests.yml index d817ca9b..e5280666 100644 --- a/.github/workflows/backend-tests.yml +++ b/.github/workflows/backend-tests.yml @@ -13,7 +13,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - lambda: [_shared, daily-fetch, loader, members, sittings, bills, member-content, member-speeches, member-speeches-publisher, member-votes, member-votes-publisher, on-this-day, estimates, riding-boundary, calendar, config, toronto-council, openapi, hansard-subjects-index, members-publisher, sittings-publisher, bills-publisher] + lambda: [_shared, daily-fetch, loader, members, sittings, bills, member-content, member-speeches, member-speeches-publisher, member-votes, member-votes-publisher, on-this-day, estimates, riding-boundary, calendar, config, toronto-council, openapi, hansard-subjects-index, hansard-search-index, members-publisher, sittings-publisher, bills-publisher] name: Test ${{ matrix.lambda }} steps: - uses: actions/checkout@v4 diff --git a/.github/workflows/hansard-search-reindex.yml b/.github/workflows/hansard-search-reindex.yml new file mode 100644 index 00000000..a75d23fc --- /dev/null +++ b/.github/workflows/hansard-search-reindex.yml @@ -0,0 +1,126 @@ +name: Hansard Search Reindex + +on: + workflow_dispatch: + inputs: + environment: + description: Environment to reindex + required: true + type: choice + options: + - staging + - production + parliament_number: + description: Optional parliament number override + required: false + type: string + session_number: + description: Optional session number override + required: false + type: string + +concurrency: + group: hansard-search-reindex-${{ inputs.environment }} + cancel-in-progress: false + +permissions: + contents: read + id-token: write + +jobs: + reindex: + runs-on: ubuntu-latest + env: + ENVIRONMENT: ${{ inputs.environment }} + FUNCTION_NAME: epac-hansard-search-index-${{ inputs.environment }} + PARLIAMENT_NUMBER: ${{ inputs.parliament_number }} + SESSION_NUMBER: ${{ inputs.session_number }} + steps: + - uses: actions/checkout@v4 + + - name: Configure AWS credentials + uses: aws-actions/configure-aws-credentials@v4 + with: + role-to-assume: ${{ inputs.environment == 'production' && secrets.AWS_BACKEND_PRODUCTION_ROLE_ARN || secrets.AWS_BACKEND_STAGING_ROLE_ARN }} + aws-region: us-east-1 + + - name: Apply optional session override + if: ${{ inputs.parliament_number != '' || inputs.session_number != '' }} + run: | + set -euo pipefail + + CURRENT_ENV=$(aws lambda get-function-configuration \ + --function-name "$FUNCTION_NAME" \ + --query 'Environment.Variables' \ + --output json) + printf '%s' "${CURRENT_ENV:-null}" > /tmp/hansard-search-index.original-env.json + UPDATED_ENV=$(jq -cn \ + --argjson current "${CURRENT_ENV:-null}" \ + --arg parliament_number "$PARLIAMENT_NUMBER" \ + --arg session_number "$SESSION_NUMBER" \ + '{Variables: (($current // {}) + (if $parliament_number != "" then {PARLIAMENT_NUMBER: $parliament_number} else {} end) + (if $session_number != "" then {SESSION_NUMBER: $session_number} else {} end))}') + + aws lambda update-function-configuration \ + --function-name "$FUNCTION_NAME" \ + --environment "$UPDATED_ENV" \ + >/dev/null + aws lambda wait function-updated \ + --function-name "$FUNCTION_NAME" + + - name: Invoke indexer + run: | + set -euo pipefail + + aws lambda invoke \ + --function-name "$FUNCTION_NAME" \ + --payload '{}' \ + --cli-binary-format raw-in-base64-out \ + /tmp/hansard-search-index.out.json \ + > /tmp/hansard-search-index.invoke.json + + { + echo "## Hansard search reindex" + echo "" + echo "- Environment: \`$ENVIRONMENT\`" + echo "- Function: \`$FUNCTION_NAME\`" + if [ -n "$PARLIAMENT_NUMBER" ]; then + echo "- Parliament override: \`$PARLIAMENT_NUMBER\`" + fi + if [ -n "$SESSION_NUMBER" ]; then + echo "- Session override: \`$SESSION_NUMBER\`" + fi + echo "" + echo "### Invoke response" + echo '```json' + jq . /tmp/hansard-search-index.invoke.json + echo '```' + echo "" + echo "### Function payload" + echo '```json' + jq . /tmp/hansard-search-index.out.json 2>/dev/null || cat /tmp/hansard-search-index.out.json + echo '```' + } >> "$GITHUB_STEP_SUMMARY" + + if jq -e '.FunctionError? // empty' /tmp/hansard-search-index.invoke.json >/dev/null; then + exit 1 + fi + + - name: Restore session override + if: ${{ always() && (inputs.parliament_number != '' || inputs.session_number != '') }} + run: | + set -euo pipefail + + if [ ! -f /tmp/hansard-search-index.original-env.json ]; then + exit 0 + fi + ORIGINAL_ENV=$(cat /tmp/hansard-search-index.original-env.json) + RESTORED_ENV=$(jq -cn \ + --argjson current "${ORIGINAL_ENV:-null}" \ + '{Variables: ($current // {})}') + + aws lambda update-function-configuration \ + --function-name "$FUNCTION_NAME" \ + --environment "$RESTORED_ENV" \ + >/dev/null + aws lambda wait function-updated \ + --function-name "$FUNCTION_NAME" diff --git a/backend/go.work b/backend/go.work index 406cc183..ce9ca807 100644 --- a/backend/go.work +++ b/backend/go.work @@ -10,6 +10,7 @@ use ( ./daily-fetch ./estimates ./hansard-backfill + ./hansard-search-index ./hansard-subjects-index ./health ./loader diff --git a/backend/hansard-search-index/.gitignore b/backend/hansard-search-index/.gitignore new file mode 100644 index 00000000..e62a59ce --- /dev/null +++ b/backend/hansard-search-index/.gitignore @@ -0,0 +1,4 @@ +bootstrap +function.zip +!testdata/ +!testdata/*.xml diff --git a/backend/hansard-search-index/README.md b/backend/hansard-search-index/README.md new file mode 100644 index 00000000..83127633 --- /dev/null +++ b/backend/hansard-search-index/README.md @@ -0,0 +1,141 @@ +# hansard-search-index + +`hansard-search-index` is a manually-invoked Go Lambda that builds the v1 +SQLite FTS5 search index for the current House of Commons Hansard session. It +downloads English Hansard XML from ourcommons.ca, parses interventions and +paragraphs, writes `/tmp/index.sqlite`, uploads it to S3, then writes a manifest +after the SQLite object exists. + +## Invocation + +Staging: + +```bash +aws lambda invoke \ + --function-name epac-hansard-search-index-staging \ + --payload '{}' \ + --cli-binary-format raw-in-base64-out \ + /tmp/hansard-search-index.out.json +``` + +Production: + +```bash +aws lambda invoke \ + --function-name epac-hansard-search-index-production \ + --payload '{}' \ + --cli-binary-format raw-in-base64-out \ + /tmp/hansard-search-index.out.json +``` + +The manual dispatch workflow is: + +```text +https://github.com/RiddimSoftware/epac/actions/workflows/hansard-search-reindex.yml +``` + +The workflow accepts `environment`, plus optional `parliament_number` and +`session_number` overrides. The Lambda reads `PARLIAMENT_NUMBER` and +`SESSION_NUMBER` from its environment; when workflow overrides are supplied, the +workflow temporarily updates those environment variables before invoking the +function, then restores the prior configuration. + +## Configuration + +Required: + +- `EPAC_ARTIFACT_BUCKET` - destination artifact bucket. + +Optional: + +- `EPAC_HANSARD_SEARCH_PREFIX` - destination prefix, default + `hansard-search/v1`. +- `PARLIAMENT_NUMBER` - parliament to index, default `45`. +- `SESSION_NUMBER` - session to index, default `1`. + +The default session values match the current 45th Parliament, 1st Session at the +time this Lambda was added. Operators should pass workflow overrides or update +the Lambda environment when a future session starts. + +## Source URL + +The downloader uses the existing `backend/download_hansard.py` convention, which +was verified against ourcommons.ca: + +```text +https://www.ourcommons.ca/Content/House/{parliament}{session}/Debates/{sitting:03d}/HAN{sitting:03d}-E.XML +``` + +For example: + +```text +https://www.ourcommons.ca/Content/House/451/Debates/001/HAN001-E.XML +``` + +Raw XML is cached for one invocation under `/tmp/raw/45-1-HAN001-E.XML` style +filenames. + +## S3 Layout + +With the default prefix, the Lambda writes: + +```text +s3://${EPAC_ARTIFACT_BUCKET}/hansard-search/v1/index.sqlite +s3://${EPAC_ARTIFACT_BUCKET}/hansard-search/v1/manifest.json +``` + +The manifest shape is: + +```json +{ + "version": "v1", + "built_at": "2026-05-25T17:30:00Z", + "parliament_number": 45, + "session_number": 1, + "sitting_count": 47, + "intervention_count": 12345, + "message_count": 56789, + "sqlite_key": "hansard-search/v1/index.sqlite", + "sqlite_size_bytes": 31415926, + "sqlite_sha256": "" +} +``` + +## SQLite Schema + +The builder recreates the database on every run: + +```sql +CREATE TABLE meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); +CREATE TABLE interventions ( + rowid INTEGER PRIMARY KEY, + parliament_number INTEGER NOT NULL, + session_number INTEGER NOT NULL, + sitting_date TEXT NOT NULL, + intervention_id TEXT NOT NULL UNIQUE, + speaker_name TEXT NOT NULL, + party_abbreviation TEXT NOT NULL DEFAULT '', + riding_name TEXT NOT NULL DEFAULT '', + topic TEXT NOT NULL DEFAULT '' +); +CREATE INDEX idx_interventions_sitting_date ON interventions(sitting_date); +CREATE TABLE messages ( + rowid INTEGER PRIMARY KEY, + intervention_rowid INTEGER NOT NULL REFERENCES interventions(rowid), + message_id TEXT NOT NULL UNIQUE, + position INTEGER NOT NULL, + content TEXT NOT NULL +); +CREATE VIRTUAL TABLE messages_fts USING fts5( + content, + content='messages', + content_rowid='rowid', + tokenize='porter unicode61 remove_diacritics 1' +); +``` + +Triggers keep `messages_fts` synchronized with `messages`. Dates are stored as +`YYYY-MM-DD`. diff --git a/backend/hansard-search-index/go.mod b/backend/hansard-search-index/go.mod new file mode 100644 index 00000000..a31fb9c7 --- /dev/null +++ b/backend/hansard-search-index/go.mod @@ -0,0 +1,46 @@ +module epac/hansard-search-index + +go 1.24.0 + +require ( + epac/observability v0.0.0 + github.com/aws/aws-lambda-go v1.54.0 + github.com/aws/aws-sdk-go-v2 v1.40.0 + github.com/aws/aws-sdk-go-v2/config v1.32.1 + github.com/aws/aws-sdk-go-v2/service/s3 v1.88.0 + modernc.org/sqlite v1.39.1 +) + +require ( + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.1 // indirect + github.com/aws/aws-sdk-go-v2/credentials v1.19.1 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.14 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.14 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.14 // indirect + github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.7 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.3 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.8.7 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.14 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.7 // indirect + github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.40.1 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.0.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.30.4 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.9 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.41.1 // indirect + github.com/aws/smithy-go v1.23.2 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect + github.com/getsentry/sentry-go v0.35.3 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/mattn/go-isatty v0.0.20 // indirect + github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect + golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b // indirect + golang.org/x/sys v0.36.0 // indirect + golang.org/x/text v0.14.0 // indirect + modernc.org/libc v1.66.10 // indirect + modernc.org/mathutil v1.7.1 // indirect + modernc.org/memory v1.11.0 // indirect +) + +replace epac/observability => ../observability diff --git a/backend/hansard-search-index/go.sum b/backend/hansard-search-index/go.sum new file mode 100644 index 00000000..0de82048 --- /dev/null +++ b/backend/hansard-search-index/go.sum @@ -0,0 +1,113 @@ +github.com/aws/aws-lambda-go v1.54.0 h1:EGYpdyRGF88xszqlGcBewz811mJeRS+maNlLZXFheII= +github.com/aws/aws-lambda-go v1.54.0/go.mod h1:dpMpZgvWx5vuQJfBt0zqBha60q7Dd7RfgJv23DymV8A= +github.com/aws/aws-sdk-go-v2 v1.40.0 h1:/WMUA0kjhZExjOQN2z3oLALDREea1A7TobfuiBrKlwc= +github.com/aws/aws-sdk-go-v2 v1.40.0/go.mod h1:c9pm7VwuW0UPxAEYGyTmyurVcNrbF6Rt/wixFqDhcjE= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.1 h1:i8p8P4diljCr60PpJp6qZXNlgX4m2yQFpYk+9ZT+J4E= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.1/go.mod h1:ddqbooRZYNoJ2dsTwOty16rM+/Aqmk/GOXrK8cg7V00= +github.com/aws/aws-sdk-go-v2/config v1.32.1 h1:iODUDLgk3q8/flEC7ymhmxjfoAnBDwEEYEVyKZ9mzjU= +github.com/aws/aws-sdk-go-v2/config v1.32.1/go.mod h1:xoAgo17AGrPpJBSLg81W+ikM0cpOZG8ad04T2r+d5P0= +github.com/aws/aws-sdk-go-v2/credentials v1.19.1 h1:JeW+EwmtTE0yXFK8SmklrFh/cGTTXsQJumgMZNlbxfM= +github.com/aws/aws-sdk-go-v2/credentials v1.19.1/go.mod h1:BOoXiStwTF+fT2XufhO0Efssbi1CNIO/ZXpZu87N0pw= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.14 h1:WZVR5DbDgxzA0BJeudId89Kmgy6DIU4ORpxwsVHz0qA= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.14/go.mod h1:Dadl9QO0kHgbrH1GRqGiZdYtW5w+IXXaBNCHTIaheM4= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.14 h1:PZHqQACxYb8mYgms4RZbhZG0a7dPW06xOjmaH0EJC/I= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.14/go.mod h1:VymhrMJUWs69D8u0/lZ7jSB6WgaG/NqHi3gX0aYf6U0= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.14 h1:bOS19y6zlJwagBfHxs0ESzr1XCOU2KXJCWcq3E2vfjY= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.14/go.mod h1:1ipeGBMAxZ0xcTm6y6paC2C/J6f6OO7LBODV9afuAyM= +github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4 h1:WKuaxf++XKWlHWu9ECbMlha8WOEGm0OUEZqm4K/Gcfk= +github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4/go.mod h1:ZWy7j6v1vWGmPReu0iSGvRiise4YI5SkR3OHKTZ6Wuc= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.7 h1:BszAktdUo2xlzmYHjWMq70DqJ7cROM8iBd3f6hrpuMQ= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.7/go.mod h1:XJ1yHki/P7ZPuG4fd3f0Pg/dSGA2cTQBCLw82MH2H48= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.3 h1:x2Ibm/Af8Fi+BH+Hsn9TXGdT+hKbDd5XOTZxTMxDk7o= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.3/go.mod h1:IW1jwyrQgMdhisceG8fQLmQIydcT/jWY21rFhzgaKwo= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.8.7 h1:zmZ8qvtE9chfhBPuKB2aQFxW5F/rpwXUgmcVCgQzqRw= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.8.7/go.mod h1:vVYfbpd2l+pKqlSIDIOgouxNsGu5il9uDp0ooWb0jys= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.14 h1:FIouAnCE46kyYqyhs0XEBDFFSREtdnr8HQuLPQPLCrY= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.14/go.mod h1:UTwDc5COa5+guonQU8qBikJo1ZJ4ln2r1MkF7Dqag1E= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.7 h1:u3VbDKUCWarWiU+aIUK4gjTr/wQFXV17y3hgNno9fcA= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.7/go.mod h1:/OuMQwhSyRapYxq6ZNpPer8juGNrB4P5Oz8bZ2cgjQE= +github.com/aws/aws-sdk-go-v2/service/s3 v1.88.0 h1:k5JXPr+2SrPDwM3PdygZUenn0lVPLa3KOs7cCYqinFs= +github.com/aws/aws-sdk-go-v2/service/s3 v1.88.0/go.mod h1:xajPTguLoeQMAOE44AAP2RQoUhF8ey1g5IFHARv71po= +github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.40.1 h1:w6a0H79HrHf3lr+zrw+pSzR5B+caiQFAKiNHlrUcnoc= +github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.40.1/go.mod h1:c6Vg0BRiU7v0MVhHupw90RyL120QBwAMLbDCzptGeMk= +github.com/aws/aws-sdk-go-v2/service/signin v1.0.1 h1:BDgIUYGEo5TkayOWv/oBLPphWwNm/A91AebUjAu5L5g= +github.com/aws/aws-sdk-go-v2/service/signin v1.0.1/go.mod h1:iS6EPmNeqCsGo+xQmXv0jIMjyYtQfnwg36zl2FwEouk= +github.com/aws/aws-sdk-go-v2/service/sso v1.30.4 h1:U//SlnkE1wOQiIImxzdY5PXat4Wq+8rlfVEw4Y7J8as= +github.com/aws/aws-sdk-go-v2/service/sso v1.30.4/go.mod h1:av+ArJpoYf3pgyrj6tcehSFW+y9/QvAY8kMooR9bZCw= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.9 h1:LU8S9W/mPDAU9q0FjCLi0TrCheLMGwzbRpvUMwYspcA= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.9/go.mod h1:/j67Z5XBVDx8nZVp9EuFM9/BS5dvBznbqILGuu73hug= +github.com/aws/aws-sdk-go-v2/service/sts v1.41.1 h1:GdGmKtG+/Krag7VfyOXV17xjTCz0i9NT+JnqLTOI5nA= +github.com/aws/aws-sdk-go-v2/service/sts v1.41.1/go.mod h1:6TxbXoDSgBQ225Qd8Q+MbxUxUh6TtNKwbRt/EPS9xso= +github.com/aws/smithy-go v1.23.2 h1:Crv0eatJUQhaManss33hS5r40CG3ZFH+21XSkqMrIUM= +github.com/aws/smithy-go v1.23.2/go.mod h1:LEj2LM3rBRQJxPZTB4KuzZkaZYnZPnvgIhb4pu07mx0= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= +github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= +github.com/getsentry/sentry-go v0.35.3 h1:u5IJaEqZyPdWqe/hKlBKBBnMTSxB/HenCqF3QLabeds= +github.com/getsentry/sentry-go v0.35.3/go.mod h1:mdL49ixwT2yi57k5eh7mpnDyPybixPzlzEJFu0Z76QA= +github.com/go-errors/errors v1.4.2 h1:J6MZopCL4uSllY1OfXM374weqZFFItUbrImctkmUxIA= +github.com/go-errors/errors v1.4.2/go.mod h1:sIVyrIiJhuEF+Pj9Ebtd6P/rEYROXFi3BopGUQ5a5Og= +github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= +github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= +github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= +github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4= +github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls= +github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4= +github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8= +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 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= +github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= +github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk= +github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= +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/exp v0.0.0-20250620022241-b7579e27df2b h1:M2rDM6z3Fhozi9O7NWsxAkg/yqS/lQJ6PmkyIV3YP+o= +golang.org/x/exp v0.0.0-20250620022241-b7579e27df2b/go.mod h1:3//PLf8L/X+8b4vuAfHzxeRUl04Adcb341+IGKfnqS8= +golang.org/x/mod v0.27.0 h1:kb+q2PyFnEADO2IEF935ehFUXlWiNjJWtRNgBLSfbxQ= +golang.org/x/mod v0.27.0/go.mod h1:rWI627Fq0DEoudcK+MBkNkCe0EetEaDSwJJkCcjpazc= +golang.org/x/sync v0.16.0 h1:ycBJEhp9p4vXvUZNszeOq0kGTPghopOL8q0fq3vstxw= +golang.org/x/sync v0.16.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.36.0 h1:KVRy2GtZBrk1cBYA7MKu5bEZFxQk4NIDV6RLVcC8o0k= +golang.org/x/sys v0.36.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ= +golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= +golang.org/x/tools v0.36.0 h1:kWS0uv/zsvHEle1LbV5LE8QujrxB3wfQyxHfhOk0Qkg= +golang.org/x/tools v0.36.0/go.mod h1:WBDiHKJK8YgLHlcQPYQzNCkUxUypCaa5ZegCVutKm+s= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +modernc.org/cc/v4 v4.26.5 h1:xM3bX7Mve6G8K8b+T11ReenJOT+BmVqQj0FY5T4+5Y4= +modernc.org/cc/v4 v4.26.5/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0= +modernc.org/ccgo/v4 v4.28.1 h1:wPKYn5EC/mYTqBO373jKjvX2n+3+aK7+sICCv4Fjy1A= +modernc.org/ccgo/v4 v4.28.1/go.mod h1:uD+4RnfrVgE6ec9NGguUNdhqzNIeeomeXf6CL0GTE5Q= +modernc.org/fileutil v1.3.40 h1:ZGMswMNc9JOCrcrakF1HrvmergNLAmxOPjizirpfqBA= +modernc.org/fileutil v1.3.40/go.mod h1:HxmghZSZVAz/LXcMNwZPA/DRrQZEVP9VX0V4LQGQFOc= +modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI= +modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito= +modernc.org/goabi0 v0.2.0 h1:HvEowk7LxcPd0eq6mVOAEMai46V+i7Jrj13t4AzuNks= +modernc.org/goabi0 v0.2.0/go.mod h1:CEFRnnJhKvWT1c1JTI3Avm+tgOWbkOu5oPA8eH8LnMI= +modernc.org/libc v1.66.10 h1:yZkb3YeLx4oynyR+iUsXsybsX4Ubx7MQlSYEw4yj59A= +modernc.org/libc v1.66.10/go.mod h1:8vGSEwvoUoltr4dlywvHqjtAqHBaw0j1jI7iFBTAr2I= +modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU= +modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg= +modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI= +modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw= +modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8= +modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns= +modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w= +modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE= +modernc.org/sqlite v1.39.1 h1:H+/wGFzuSCIEVCvXYVHX5RQglwhMOvtHSv+VtidL2r4= +modernc.org/sqlite v1.39.1/go.mod h1:9fjQZ0mB1LLP0GYrp39oOJXx/I2sxEnZtzCmEQIKvGE= +modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0= +modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A= +modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y= +modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM= diff --git a/backend/hansard-search-index/internal/adapter/ourcommons/log.go b/backend/hansard-search-index/internal/adapter/ourcommons/log.go new file mode 100644 index 00000000..a1316390 --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/ourcommons/log.go @@ -0,0 +1,67 @@ +package ourcommons + +import ( + "encoding/json" + "io" + "os" + "sync" + "time" +) + +const pipelineName = "hansard-search-index" + +type Logger interface { + Info(event string, fields map[string]any) + Warn(event string, fields map[string]any) +} + +type JSONLogger struct { + out io.Writer + mu sync.Mutex +} + +func NewJSONLogger(out io.Writer) *JSONLogger { + if out == nil { + out = os.Stdout + } + return &JSONLogger{out: out} +} + +func (l *JSONLogger) Info(event string, fields map[string]any) { + l.write("info", event, fields) +} + +func (l *JSONLogger) Warn(event string, fields map[string]any) { + l.write("warn", event, fields) +} + +func (l *JSONLogger) write(level, event string, fields map[string]any) { + record := map[string]any{ + "timestamp": time.Now().UTC().Format(time.RFC3339), + "level": level, + "pipeline": pipelineName, + "event": event, + } + for key, value := range fields { + record[key] = value + } + data, err := json.Marshal(record) + if err != nil { + return + } + l.mu.Lock() + defer l.mu.Unlock() + _, _ = l.out.Write(append(data, '\n')) +} + +type discardLogger struct{} + +func (discardLogger) Info(string, map[string]any) {} +func (discardLogger) Warn(string, map[string]any) {} + +func defaultLogger(logger Logger) Logger { + if logger == nil { + return discardLogger{} + } + return logger +} diff --git a/backend/hansard-search-index/internal/adapter/ourcommons/parser.go b/backend/hansard-search-index/internal/adapter/ourcommons/parser.go new file mode 100644 index 00000000..ee3f14f6 --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/ourcommons/parser.go @@ -0,0 +1,319 @@ +package ourcommons + +import ( + "bytes" + "encoding/xml" + "fmt" + "io" + "regexp" + "strconv" + "strings" + "time" + + "epac/hansard-search-index/internal/domain" +) + +type Parser struct { + logger Logger + unknownSeen map[string]bool +} + +func NewParser(logger Logger) *Parser { + return &Parser{ + logger: defaultLogger(logger), + unknownSeen: map[string]bool{}, + } +} + +func (p *Parser) Parse(body []byte) ([]domain.Intervention, error) { + decoder := xml.NewDecoder(bytes.NewReader(body)) + var interventions []domain.Intervention + + var parliamentNumber int + var sessionNumber int + var sittingDate time.Time + var inExtractedItem bool + var currentItemName string + + var currentSubject string + var inSubjectTitle bool + var current *interventionBuilder + var inPersonSpeaking int + var inContent int + var captureSpeaker bool + var inParaText int + + for { + tok, err := decoder.Token() + if err == io.EOF { + break + } + if err != nil { + p.warn("malformed_xml", map[string]any{"error": err.Error()}) + return nil, fmt.Errorf("decode XML: %w", err) + } + + switch se := tok.(type) { + case xml.StartElement: + p.noteUnknown(se) + switch se.Name.Local { + case "ExtractedItem": + inExtractedItem = true + currentItemName = attrValue(se, "Name") + case "SubjectOfBusiness": + currentSubject = "" + case "SubjectOfBusinessTitle": + inSubjectTitle = true + case "Intervention": + current = &interventionBuilder{ + Intervention: domain.Intervention{ + ParliamentNumber: parliamentNumber, + SessionNumber: sessionNumber, + SittingDate: sittingDate, + InterventionID: strings.TrimSpace(attrValue(se, "id")), + Topic: strings.TrimSpace(currentSubject), + }, + } + case "PersonSpeaking": + inPersonSpeaking++ + case "Content": + inContent++ + case "Affiliation": + if current != nil && inPersonSpeaking > 0 && inContent == 0 { + captureSpeaker = true + } + case "ParaText": + if current != nil && inContent > 0 { + inParaText++ + id := strings.TrimSpace(attrValue(se, "id")) + if id == "" { + p.warn("malformed_para_text", map[string]any{ + "intervention_id": current.InterventionID, + "reason": "missing id", + }) + continue + } + current.startMessage(id) + } + } + case xml.CharData: + text := string(se) + switch { + case inExtractedItem: + switch currentItemName { + case "ParliamentNumber": + parliamentNumber, _ = strconv.Atoi(strings.TrimSpace(text)) + case "SessionNumber": + sessionNumber, _ = strconv.Atoi(strings.TrimSpace(text)) + case "Date": + sittingDate = parseHansardDate(strings.TrimSpace(text)) + } + case inSubjectTitle && current == nil: + currentSubject += text + case current != nil && captureSpeaker: + current.speaker += text + case current != nil && inParaText > 0: + current.writeMessageText(text) + } + case xml.EndElement: + switch se.Name.Local { + case "ExtractedItem": + inExtractedItem = false + currentItemName = "" + case "SubjectOfBusinessTitle": + inSubjectTitle = false + currentSubject = normalizeWhitespace(currentSubject) + case "Intervention": + if current != nil { + intervention, ok := current.finish(p.logger) + if ok { + interventions = append(interventions, intervention) + } + current = nil + } + case "PersonSpeaking": + if inPersonSpeaking > 0 { + inPersonSpeaking-- + } + case "Affiliation": + captureSpeaker = false + case "Content": + if inContent > 0 { + inContent-- + } + case "ParaText": + if current != nil && inParaText > 0 { + inParaText-- + current.endMessage() + } + } + } + } + return interventions, nil +} + +type interventionBuilder struct { + domain.Intervention + speaker string + currentMessage *domain.Message +} + +func (b *interventionBuilder) startMessage(id string) { + b.currentMessage = &domain.Message{ + MessageID: id, + Position: len(b.Messages) + 1, + } +} + +func (b *interventionBuilder) writeMessageText(text string) { + if b.currentMessage != nil { + b.currentMessage.Text += text + } +} + +func (b *interventionBuilder) endMessage() { + if b.currentMessage == nil { + return + } + b.currentMessage.Text = normalizeWhitespace(b.currentMessage.Text) + if b.currentMessage.Text != "" { + b.Messages = append(b.Messages, *b.currentMessage) + } + b.currentMessage = nil +} + +func (b *interventionBuilder) finish(logger Logger) (domain.Intervention, bool) { + if b.InterventionID == "" { + logger.Warn("malformed_intervention", map[string]any{"reason": "missing id"}) + return domain.Intervention{}, false + } + speaker := parseSpeaker(b.speaker) + b.SpeakerFirstName = speaker.firstName + b.SpeakerLastName = speaker.lastName + b.PartyAbbreviation = speaker.partyAbbreviation + b.RidingName = speaker.ridingName + return b.Intervention, true +} + +type parsedSpeaker struct { + firstName string + lastName string + partyAbbreviation string + ridingName string +} + +func parseSpeaker(text string) parsedSpeaker { + text = strings.TrimSpace(strings.ReplaceAll(text, "Mme ", "Mme. ")) + name := strings.TrimSpace(text) + details := "" + if open := strings.Index(text, "("); open >= 0 { + name = strings.TrimSpace(text[:open]) + if close := strings.LastIndex(text, ")"); close > open { + details = strings.TrimSpace(text[open+1 : close]) + } + } + party := partyFromDetails(details) + riding := ridingFromDetails(details) + if party == "" && riding == "" && details != "" { + name = details + } + first, last := speakerNameParts(name) + return parsedSpeaker{ + firstName: first, + lastName: last, + partyAbbreviation: party, + ridingName: riding, + } +} + +func partyFromDetails(details string) string { + comma := strings.LastIndex(details, ",") + if comma == -1 { + return "" + } + return strings.TrimSuffix(strings.TrimSpace(details[comma+1:]), ".") +} + +func ridingFromDetails(details string) string { + comma := strings.LastIndex(details, ",") + if comma == -1 { + return "" + } + ridingOrRole := strings.TrimSpace(details[:comma]) + if roleComma := strings.LastIndex(ridingOrRole, ","); roleComma >= 0 { + return strings.TrimSpace(ridingOrRole[roleComma+1:]) + } + if strings.Contains(ridingOrRole, "(") { + return "" + } + return ridingOrRole +} + +func speakerNameParts(name string) (string, string) { + parts := strings.Fields(name) + cleaned := make([]string, 0, len(parts)) + for _, part := range parts { + if !speakerTitleWords[part] { + cleaned = append(cleaned, part) + } + } + if len(cleaned) == 0 { + return "", "" + } + if len(cleaned) == 1 { + return "", cleaned[0] + } + return strings.Join(cleaned[:len(cleaned)-1], " "), cleaned[len(cleaned)-1] +} + +var speakerTitleWords = map[string]bool{ + "Hon.": true, "Rt.": true, "Mr.": true, "Ms.": true, "Mrs.": true, + "Mme.": true, "Mme": true, "M.": true, "L'hon.": true, "L'hon": true, + "Dr.": true, "The": true, "Hon": true, "Rt": true, "Right": true, +} + +func attrValue(se xml.StartElement, name string) string { + for _, attr := range se.Attr { + if attr.Name.Local == name { + return attr.Value + } + } + return "" +} + +func parseHansardDate(s string) time.Time { + dayPrefix := regexp.MustCompile(`^(Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday),\s+`) + cleaned := dayPrefix.ReplaceAllString(strings.TrimSpace(s), "") + t, err := time.Parse("January 2, 2006", cleaned) + if err != nil { + return time.Time{} + } + return t +} + +func normalizeWhitespace(s string) string { + return strings.Join(strings.Fields(s), " ") +} + +func (p *Parser) noteUnknown(se xml.StartElement) { + if knownElements[se.Name.Local] || p.unknownSeen[se.Name.Local] { + return + } + p.unknownSeen[se.Name.Local] = true + p.warn("unknown_element", map[string]any{"element": se.Name.Local}) +} + +func (p *Parser) warn(event string, fields map[string]any) { + p.logger.Warn(event, fields) +} + +var knownElements = map[string]bool{ + "Affiliation": true, "CatchLine": true, "Content": true, "DocumentTitle": true, + "ExtractedInformation": true, "ExtractedItem": true, "FloorLanguage": true, + "Hansard": true, "HansardBody": true, "Intervention": true, "Intro": true, + "I": true, "OrderOfBusiness": true, "OrderOfBusinessTitle": true, "ParaText": true, + "PersonSpeaking": true, "Prayer": true, "ProceduralText": true, "Quote": true, + "QuotePara": true, "StartPageNumber": true, "SubjectOfBusiness": true, + "SubjectOfBusinessContent": true, "SubjectOfBusinessQualifier": true, + "SubjectOfBusinessTitle": true, "Sup": true, "Timestamp": true, +} diff --git a/backend/hansard-search-index/internal/adapter/ourcommons/parser_test.go b/backend/hansard-search-index/internal/adapter/ourcommons/parser_test.go new file mode 100644 index 00000000..91cd21fd --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/ourcommons/parser_test.go @@ -0,0 +1,69 @@ +package ourcommons + +import ( + "os" + "path/filepath" + "testing" +) + +func TestParserExtractsInterventionsFromRealHansardFixture(t *testing.T) { + fixture := filepath.Join("..", "..", "..", "testdata", "hansard_451_001_slice.xml") + body, err := os.ReadFile(fixture) + if err != nil { + t.Fatalf("read fixture: %v", err) + } + + got, err := NewParser(nil).Parse(body) + if err != nil { + t.Fatalf("Parse: %v", err) + } + if len(got) != 2 { + t.Fatalf("intervention count = %d, want 2", len(got)) + } + + first := got[0] + if first.ParliamentNumber != 45 || first.SessionNumber != 1 { + t.Fatalf("session = %d-%d, want 45-1", first.ParliamentNumber, first.SessionNumber) + } + if gotDate := first.SittingDate.Format("2006-01-02"); gotDate != "2025-05-26" { + t.Fatalf("sitting date = %q", gotDate) + } + if first.InterventionID != "13077673" { + t.Fatalf("intervention id = %q", first.InterventionID) + } + if first.SpeakerFirstName != "John" || first.SpeakerLastName != "Nater" { + t.Fatalf("speaker = %q %q, want John Nater", first.SpeakerFirstName, first.SpeakerLastName) + } + if first.PartyAbbreviation != "CPC" { + t.Fatalf("party = %q, want CPC", first.PartyAbbreviation) + } + if first.RidingName != "Perth—Wellington" { + t.Fatalf("riding = %q, want Perth—Wellington", first.RidingName) + } + if first.Topic != "Election of Speaker" { + t.Fatalf("topic = %q, want Election of Speaker", first.Topic) + } + if len(first.Messages) != 1 { + t.Fatalf("messages = %d, want 1", len(first.Messages)) + } + if first.Messages[0].MessageID != "8830321" || first.Messages[0].Position != 1 { + t.Fatalf("message metadata = %#v", first.Messages[0]) + } + if first.Messages[0].Text == "" { + t.Fatal("message text must not be empty") + } + + second := got[1] + if second.InterventionID != "13077675" { + t.Fatalf("second intervention id = %q", second.InterventionID) + } + if second.SpeakerFirstName != "Chris" || second.SpeakerLastName != "d'Entremont" { + t.Fatalf("second speaker = %q %q", second.SpeakerFirstName, second.SpeakerLastName) + } + if len(second.Messages) != 2 { + t.Fatalf("second messages = %d, want 2", len(second.Messages)) + } + if second.Messages[1].MessageID != "8830323" || second.Messages[1].Position != 2 { + t.Fatalf("second message metadata = %#v", second.Messages[1]) + } +} diff --git a/backend/hansard-search-index/internal/adapter/ourcommons/source.go b/backend/hansard-search-index/internal/adapter/ourcommons/source.go new file mode 100644 index 00000000..bc486c63 --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/ourcommons/source.go @@ -0,0 +1,234 @@ +package ourcommons + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "epac/hansard-search-index/internal/usecase" +) + +const ( + DefaultBaseURL = "https://www.ourcommons.ca/Content/House" + DefaultCacheDir = "/tmp/raw" + DefaultUserAgent = "epac-hansard-search-index/1.0 (sunny@riddimsoftware.com)" + MaxRetries = 4 + MaxBackoff = 30 * time.Second +) + +type Source struct { + client *http.Client + baseURL string + cacheDir string + userAgent string + logger Logger + sleep func(context.Context, time.Duration) error +} + +type SourceOption func(*Source) + +func NewSource(options ...SourceOption) *Source { + source := &Source{ + client: &http.Client{Timeout: 30 * time.Second}, + baseURL: DefaultBaseURL, + cacheDir: DefaultCacheDir, + userAgent: DefaultUserAgent, + logger: discardLogger{}, + sleep: sleepContext, + } + for _, option := range options { + option(source) + } + source.baseURL = strings.TrimRight(source.baseURL, "/") + source.logger = defaultLogger(source.logger) + return source +} + +func WithHTTPClient(client *http.Client) SourceOption { + return func(source *Source) { + if client != nil { + source.client = client + } + } +} + +func WithBaseURL(baseURL string) SourceOption { + return func(source *Source) { + if strings.TrimSpace(baseURL) != "" { + source.baseURL = baseURL + } + } +} + +func WithCacheDir(cacheDir string) SourceOption { + return func(source *Source) { + if strings.TrimSpace(cacheDir) != "" { + source.cacheDir = cacheDir + } + } +} + +func WithLogger(logger Logger) SourceOption { + return func(source *Source) { + source.logger = logger + } +} + +func WithSleep(sleep func(context.Context, time.Duration) error) SourceOption { + return func(source *Source) { + if sleep != nil { + source.sleep = sleep + } + } +} + +func (s *Source) FetchSitting(ctx context.Context, parliament, session, sitting int) ([]byte, error) { + start := time.Now() + filename := Filename(parliament, session, sitting) + cachePath := filepath.Join(s.cacheDir, filename) + if body, err := os.ReadFile(cachePath); err == nil { + s.logFetch("cache_hit", sitting, http.StatusOK, start) + return body, nil + } else if !os.IsNotExist(err) { + return nil, fmt.Errorf("read cached XML %s: %w", cachePath, err) + } + + url := URL(s.baseURL, parliament, session, sitting) + var lastStatus int + var lastErr error + for attempt := 0; attempt <= MaxRetries; attempt++ { + body, status, retryAfter, err := s.fetch(ctx, url) + lastStatus = status + lastErr = err + if err != nil { + if attempt == MaxRetries { + break + } + if err := s.sleep(ctx, backoff(attempt, retryAfter)); err != nil { + return nil, err + } + continue + } + + s.logFetch("fetch", sitting, status, start) + switch status { + case http.StatusOK: + if err := os.MkdirAll(s.cacheDir, 0o755); err != nil { + return nil, fmt.Errorf("create raw XML cache dir: %w", err) + } + if err := os.WriteFile(cachePath, body, 0o644); err != nil { + return nil, fmt.Errorf("write cached XML %s: %w", cachePath, err) + } + return body, nil + case http.StatusNotFound: + return nil, usecase.ErrSittingNotFound + case http.StatusTooManyRequests: + if attempt == MaxRetries { + return nil, fmt.Errorf("%s returned 429 after retries", url) + } + if err := s.sleep(ctx, backoff(attempt, retryAfter)); err != nil { + return nil, err + } + default: + return nil, fmt.Errorf("%s returned status %d", url, status) + } + } + if lastErr != nil { + return nil, lastErr + } + return nil, fmt.Errorf("%s returned status %d", url, lastStatus) +} + +func (s *Source) fetch(ctx context.Context, url string) ([]byte, int, time.Duration, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, 0, 0, err + } + req.Header.Set("User-Agent", s.userAgent) + resp, err := s.client.Do(req) + if err != nil { + return nil, 0, 0, err + } + defer resp.Body.Close() + + body, err := io.ReadAll(resp.Body) + if err != nil { + return nil, resp.StatusCode, retryAfter(resp.Header.Get("Retry-After")), err + } + if resp.StatusCode == http.StatusOK && !looksLikeXML(resp.Header.Get("Content-Type"), body) { + return nil, http.StatusNotFound, 0, nil + } + return body, resp.StatusCode, retryAfter(resp.Header.Get("Retry-After")), nil +} + +func (s *Source) logFetch(event string, sitting, status int, start time.Time) { + s.logger.Info(event, map[string]any{ + "sitting": sitting, + "status_code": status, + "duration_ms": time.Since(start).Milliseconds(), + }) +} + +func Filename(parliament, session, sitting int) string { + return fmt.Sprintf("%d-%d-HAN%03d-E.XML", parliament, session, sitting) +} + +func URL(baseURL string, parliament, session, sitting int) string { + return fmt.Sprintf("%s/%d%d/Debates/%03d/HAN%03d-E.XML", strings.TrimRight(baseURL, "/"), parliament, session, sitting, sitting) +} + +func looksLikeXML(contentType string, body []byte) bool { + if strings.Contains(strings.ToLower(contentType), "xml") { + return true + } + return bytes.Contains(bytes.TrimSpace(body[:min(len(body), 128)]), []byte(" delay { + delay = retryAfter + } + if delay > MaxBackoff { + return MaxBackoff + } + if delay < 0 { + return 0 + } + return delay +} + +func sleepContext(ctx context.Context, delay time.Duration) error { + if delay <= 0 { + return nil + } + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/backend/hansard-search-index/internal/adapter/ourcommons/source_test.go b/backend/hansard-search-index/internal/adapter/ourcommons/source_test.go new file mode 100644 index 00000000..26695de6 --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/ourcommons/source_test.go @@ -0,0 +1,114 @@ +package ourcommons + +import ( + "context" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" + + "epac/hansard-search-index/internal/usecase" +) + +func TestURLMatchesLiveOurCommonsPattern(t *testing.T) { + got := URL(DefaultBaseURL, 45, 1, 1) + want := "https://www.ourcommons.ca/Content/House/451/Debates/001/HAN001-E.XML" + if got != want { + t.Fatalf("URL() = %q, want %q", got, want) + } +} + +func TestFilenameUsesSessionBreadcrumb(t *testing.T) { + got := Filename(45, 1, 7) + want := "45-1-HAN007-E.XML" + if got != want { + t.Fatalf("Filename() = %q, want %q", got, want) + } +} + +func TestFetchSittingDownloadsAndCachesXML(t *testing.T) { + requests := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + if r.URL.Path != "/451/Debates/001/HAN001-E.XML" { + t.Fatalf("path = %q", r.URL.Path) + } + if got := r.Header.Get("User-Agent"); got != DefaultUserAgent { + t.Fatalf("user agent = %q", got) + } + w.Header().Set("Content-Type", "text/xml") + _, _ = w.Write([]byte(``)) + })) + defer server.Close() + + source := NewSource( + WithBaseURL(server.URL), + WithCacheDir(t.TempDir()), + WithHTTPClient(server.Client()), + WithSleep(func(context.Context, time.Duration) error { return nil }), + ) + body, err := source.FetchSitting(context.Background(), 45, 1, 1) + if err != nil { + t.Fatalf("first FetchSitting: %v", err) + } + if string(body) == "" { + t.Fatal("first body is empty") + } + body, err = source.FetchSitting(context.Background(), 45, 1, 1) + if err != nil { + t.Fatalf("cached FetchSitting: %v", err) + } + if string(body) == "" { + t.Fatal("cached body is empty") + } + if requests != 1 { + t.Fatalf("requests = %d, want 1 because second read should use cache", requests) + } +} + +func TestFetchSittingMaps404ToUseCaseNotFound(t *testing.T) { + server := httptest.NewServer(http.NotFoundHandler()) + defer server.Close() + source := NewSource(WithBaseURL(server.URL), WithCacheDir(t.TempDir()), WithHTTPClient(server.Client())) + + _, err := source.FetchSitting(context.Background(), 45, 1, 999) + if err != usecase.ErrSittingNotFound { + t.Fatalf("error = %v, want ErrSittingNotFound", err) + } +} + +func TestFetchSittingRetries429WithRetryAfter(t *testing.T) { + requests := 0 + var slept []time.Duration + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests++ + if requests == 1 { + w.Header().Set("Retry-After", "2") + http.Error(w, "rate limited", http.StatusTooManyRequests) + return + } + w.Header().Set("Content-Type", "text/xml") + _, _ = w.Write([]byte(``)) + })) + defer server.Close() + source := NewSource( + WithBaseURL(server.URL), + WithCacheDir(filepath.Join(t.TempDir(), "raw")), + WithHTTPClient(server.Client()), + WithSleep(func(_ context.Context, d time.Duration) error { + slept = append(slept, d) + return nil + }), + ) + + if _, err := source.FetchSitting(context.Background(), 45, 1, 1); err != nil { + t.Fatalf("FetchSitting: %v", err) + } + if requests != 2 { + t.Fatalf("requests = %d, want 2", requests) + } + if len(slept) != 1 || slept[0] != 2*time.Second { + t.Fatalf("slept = %v, want [2s]", slept) + } +} diff --git a/backend/hansard-search-index/internal/adapter/s3/s3.go b/backend/hansard-search-index/internal/adapter/s3/s3.go new file mode 100644 index 00000000..8c88c0ed --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/s3/s3.go @@ -0,0 +1,114 @@ +package s3 + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "os" + "path" + "strings" + + "epac/hansard-search-index/internal/adapter/sqlitefts5" + "epac/hansard-search-index/internal/domain" + + "github.com/aws/aws-sdk-go-v2/aws" + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" +) + +const ( + SQLiteContentType = "application/x-sqlite3" + ManifestContentType = "application/json" + sha256MetaKey = "content-hash-sha256" +) + +type PutObjectClient interface { + PutObject(ctx context.Context, params *awss3.PutObjectInput, optFns ...func(*awss3.Options)) (*awss3.PutObjectOutput, error) +} + +type Store struct { + client PutObjectClient + bucket string + prefix string +} + +func NewStore(client PutObjectClient, bucket, prefix string) *Store { + return &Store{ + client: client, + bucket: strings.TrimSpace(bucket), + prefix: cleanPrefix(prefix), + } +} + +func (s *Store) Upload(ctx context.Context, localPath, s3Key string) (string, int64, error) { + if s.client == nil { + return "", 0, fmt.Errorf("s3 client is required") + } + if s.bucket == "" { + return "", 0, fmt.Errorf("EPAC_ARTIFACT_BUCKET is required") + } + hash, err := sqlitefts5.SHA256File(localPath) + if err != nil { + return "", 0, err + } + file, err := os.Open(localPath) + if err != nil { + return "", 0, fmt.Errorf("open sqlite for upload: %w", err) + } + defer file.Close() + info, err := file.Stat() + if err != nil { + return "", 0, fmt.Errorf("stat sqlite for upload: %w", err) + } + _, err = s.client.PutObject(ctx, &awss3.PutObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(cleanKey(s3Key)), + Body: file, + ContentType: aws.String(SQLiteContentType), + ServerSideEncryption: types.ServerSideEncryptionAes256, + Metadata: map[string]string{ + sha256MetaKey: hash, + }, + }) + if err != nil { + return "", 0, fmt.Errorf("put sqlite object: %w", err) + } + return hash, info.Size(), nil +} + +func (s *Store) Write(ctx context.Context, manifest domain.Manifest) error { + if s.client == nil { + return fmt.Errorf("s3 client is required") + } + if s.bucket == "" { + return fmt.Errorf("EPAC_ARTIFACT_BUCKET is required") + } + data, err := json.MarshalIndent(manifest, "", " ") + if err != nil { + return fmt.Errorf("marshal manifest: %w", err) + } + _, err = s.client.PutObject(ctx, &awss3.PutObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(path.Join(s.prefix, "manifest.json")), + Body: bytes.NewReader(data), + ContentType: aws.String(ManifestContentType), + ServerSideEncryption: types.ServerSideEncryptionAes256, + }) + if err != nil { + return fmt.Errorf("put manifest object: %w", err) + } + return nil +} + +func cleanPrefix(prefix string) string { + prefix = strings.Trim(strings.TrimSpace(prefix), "/") + if prefix == "" || prefix == "." { + return "hansard-search/v1" + } + return prefix +} + +func cleanKey(key string) string { + return strings.TrimLeft(path.Clean("/"+strings.TrimSpace(key)), "/") +} diff --git a/backend/hansard-search-index/internal/adapter/s3/s3_test.go b/backend/hansard-search-index/internal/adapter/s3/s3_test.go new file mode 100644 index 00000000..359bc30f --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/s3/s3_test.go @@ -0,0 +1,76 @@ +package s3 + +import ( + "context" + "io" + "os" + "path/filepath" + "strings" + "testing" + + "epac/hansard-search-index/internal/domain" + + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" +) + +func TestUploadWritesSQLiteWithContentTypeEncryptionAndHashMetadata(t *testing.T) { + path := filepath.Join(t.TempDir(), "index.sqlite") + if err := os.WriteFile(path, []byte("sqlite bytes"), 0o644); err != nil { + t.Fatalf("write sqlite: %v", err) + } + client := &mockPutObjectClient{} + store := NewStore(client, "bucket", "hansard-search/v1") + + hash, size, err := store.Upload(context.Background(), path, "hansard-search/v1/index.sqlite") + if err != nil { + t.Fatalf("Upload: %v", err) + } + if size != int64(len("sqlite bytes")) { + t.Fatalf("size = %d", size) + } + if hash == "" || len(hash) != 64 { + t.Fatalf("hash = %q", hash) + } + got := client.inputs[0] + if got.ContentType == nil || *got.ContentType != SQLiteContentType { + t.Fatalf("content type = %v", got.ContentType) + } + if got.Metadata[sha256MetaKey] != hash { + t.Fatalf("metadata hash = %q", got.Metadata[sha256MetaKey]) + } +} + +func TestWriteUploadsManifestAtPrefixAfterMarshal(t *testing.T) { + client := &mockPutObjectClient{} + store := NewStore(client, "bucket", "hansard-search/v1") + + err := store.Write(context.Background(), domain.Manifest{ + Version: "v1", + BuiltAt: "2026-05-25T17:30:00Z", + SQLiteKey: "hansard-search/v1/index.sqlite", + SQLiteSHA256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + }) + if err != nil { + t.Fatalf("Write: %v", err) + } + got := client.inputs[0] + if got.Key == nil || *got.Key != "hansard-search/v1/manifest.json" { + t.Fatalf("key = %v", got.Key) + } + body, err := io.ReadAll(got.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + if !strings.Contains(string(body), `"sqlite_key": "hansard-search/v1/index.sqlite"`) { + t.Fatalf("manifest body missing sqlite key: %s", body) + } +} + +type mockPutObjectClient struct { + inputs []*awss3.PutObjectInput +} + +func (m *mockPutObjectClient) PutObject(_ context.Context, input *awss3.PutObjectInput, _ ...func(*awss3.Options)) (*awss3.PutObjectOutput, error) { + m.inputs = append(m.inputs, input) + return &awss3.PutObjectOutput{}, nil +} diff --git a/backend/hansard-search-index/internal/adapter/sqlitefts5/builder.go b/backend/hansard-search-index/internal/adapter/sqlitefts5/builder.go new file mode 100644 index 00000000..b8db42a6 --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/sqlitefts5/builder.go @@ -0,0 +1,342 @@ +package sqlitefts5 + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "fmt" + "io" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + "time" + + "epac/hansard-search-index/internal/domain" + + _ "modernc.org/sqlite" +) + +const DefaultPath = "/tmp/index.sqlite" + +type Clock interface { + Now() time.Time +} + +type SystemClock struct{} + +func (SystemClock) Now() time.Time { return time.Now().UTC() } + +type Builder struct { + path string + clock Clock +} + +func NewBuilder(path string, clock Clock) *Builder { + if strings.TrimSpace(path) == "" { + path = DefaultPath + } + if clock == nil { + clock = SystemClock{} + } + return &Builder{path: path, clock: clock} +} + +func (b *Builder) Build(ctx context.Context, interventions []domain.Intervention) (string, domain.Stats, error) { + if err := removeSQLiteFiles(b.path); err != nil { + return "", domain.Stats{}, err + } + if err := os.MkdirAll(filepath.Dir(b.path), 0o755); err != nil { + return "", domain.Stats{}, fmt.Errorf("create sqlite dir: %w", err) + } + + db, err := sql.Open("sqlite", b.path) + if err != nil { + return "", domain.Stats{}, fmt.Errorf("open sqlite: %w", err) + } + defer func() { + if db != nil { + _ = db.Close() + } + }() + if _, err := db.ExecContext(ctx, "PRAGMA foreign_keys=ON"); err != nil { + return "", domain.Stats{}, fmt.Errorf("enable foreign keys: %w", err) + } + if err := createSchema(ctx, db); err != nil { + return "", domain.Stats{}, err + } + + stats, err := b.insertCorpus(ctx, db, interventions) + if err != nil { + return "", domain.Stats{}, err + } + if err := writeMeta(ctx, db, stats); err != nil { + return "", domain.Stats{}, err + } + if err := selfCheck(ctx, db, stats.MessageCount, interventions); err != nil { + return "", domain.Stats{}, err + } + if err := db.Close(); err != nil { + return "", domain.Stats{}, fmt.Errorf("close sqlite: %w", err) + } + db = nil + if _, err := SHA256File(b.path); err != nil { + return "", domain.Stats{}, err + } + return b.path, stats, nil +} + +func createSchema(ctx context.Context, db *sql.DB) error { + _, err := db.ExecContext(ctx, schemaSQL) + if err != nil { + return fmt.Errorf("create schema: %w", err) + } + return nil +} + +const schemaSQL = ` +DROP TRIGGER IF EXISTS messages_ai; +DROP TRIGGER IF EXISTS messages_ad; +DROP TRIGGER IF EXISTS messages_au; +DROP TABLE IF EXISTS messages_fts; +DROP TABLE IF EXISTS messages; +DROP TABLE IF EXISTS interventions; +DROP TABLE IF EXISTS meta; + +CREATE TABLE meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); +CREATE TABLE interventions ( + rowid INTEGER PRIMARY KEY, + parliament_number INTEGER NOT NULL, + session_number INTEGER NOT NULL, + sitting_date TEXT NOT NULL, + intervention_id TEXT NOT NULL UNIQUE, + speaker_name TEXT NOT NULL, + party_abbreviation TEXT NOT NULL DEFAULT '', + riding_name TEXT NOT NULL DEFAULT '', + topic TEXT NOT NULL DEFAULT '' +); +CREATE INDEX idx_interventions_sitting_date ON interventions(sitting_date); +CREATE TABLE messages ( + rowid INTEGER PRIMARY KEY, + intervention_rowid INTEGER NOT NULL REFERENCES interventions(rowid), + message_id TEXT NOT NULL UNIQUE, + position INTEGER NOT NULL, + content TEXT NOT NULL +); +CREATE VIRTUAL TABLE messages_fts USING fts5( + content, + content='messages', + content_rowid='rowid', + tokenize='porter unicode61 remove_diacritics 1' +); +CREATE TRIGGER messages_ai AFTER INSERT ON messages BEGIN + INSERT INTO messages_fts(rowid, content) VALUES (new.rowid, new.content); +END; +CREATE TRIGGER messages_ad AFTER DELETE ON messages BEGIN + INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.rowid, old.content); +END; +CREATE TRIGGER messages_au AFTER UPDATE ON messages BEGIN + INSERT INTO messages_fts(messages_fts, rowid, content) VALUES('delete', old.rowid, old.content); + INSERT INTO messages_fts(rowid, content) VALUES (new.rowid, new.content); +END; +` + +func (b *Builder) insertCorpus(ctx context.Context, db *sql.DB, interventions []domain.Intervention) (domain.Stats, error) { + stats := domain.Stats{BuiltAt: b.clock.Now().UTC()} + if len(interventions) == 0 { + return stats, nil + } + + ordered := append([]domain.Intervention(nil), interventions...) + sort.SliceStable(ordered, func(i, j int) bool { + return sittingKey(ordered[i]) < sittingKey(ordered[j]) + }) + stats.ParliamentNumber = ordered[0].ParliamentNumber + stats.SessionNumber = ordered[0].SessionNumber + + for start := 0; start < len(ordered); { + end := start + 1 + key := sittingKey(ordered[start]) + for end < len(ordered) && sittingKey(ordered[end]) == key { + end++ + } + if err := insertSitting(ctx, db, ordered[start:end], &stats); err != nil { + return domain.Stats{}, err + } + stats.SittingCount++ + start = end + } + return stats, nil +} + +func insertSitting(ctx context.Context, db *sql.DB, interventions []domain.Intervention, stats *domain.Stats) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin sitting transaction: %w", err) + } + defer tx.Rollback() + + for _, intervention := range interventions { + if strings.TrimSpace(intervention.InterventionID) == "" { + continue + } + res, err := tx.ExecContext(ctx, ` +INSERT INTO interventions ( + parliament_number, session_number, sitting_date, intervention_id, + speaker_name, party_abbreviation, riding_name, topic +) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + intervention.ParliamentNumber, + intervention.SessionNumber, + formatDate(intervention.SittingDate), + strings.TrimSpace(intervention.InterventionID), + speakerName(intervention), + strings.TrimSpace(intervention.PartyAbbreviation), + strings.TrimSpace(intervention.RidingName), + strings.TrimSpace(intervention.Topic), + ) + if err != nil { + return fmt.Errorf("insert intervention %q: %w", intervention.InterventionID, err) + } + interventionRowID, err := res.LastInsertId() + if err != nil { + return fmt.Errorf("read intervention rowid: %w", err) + } + stats.InterventionCount++ + + for _, message := range intervention.Messages { + if strings.TrimSpace(message.MessageID) == "" || strings.TrimSpace(message.Text) == "" { + continue + } + if _, err := tx.ExecContext(ctx, ` +INSERT INTO messages (intervention_rowid, message_id, position, content) +VALUES (?, ?, ?, ?)`, + interventionRowID, + strings.TrimSpace(message.MessageID), + message.Position, + strings.TrimSpace(message.Text), + ); err != nil { + return fmt.Errorf("insert message %q: %w", message.MessageID, err) + } + stats.MessageCount++ + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit sitting transaction: %w", err) + } + return nil +} + +func writeMeta(ctx context.Context, db *sql.DB, stats domain.Stats) error { + values := map[string]string{ + "version": domain.ManifestVersion, + "built_at": stats.BuiltAt.UTC().Format(time.RFC3339), + "parliament_number": fmt.Sprintf("%d", stats.ParliamentNumber), + "session_number": fmt.Sprintf("%d", stats.SessionNumber), + "sitting_count": fmt.Sprintf("%d", stats.SittingCount), + "intervention_count": fmt.Sprintf("%d", stats.InterventionCount), + "message_count": fmt.Sprintf("%d", stats.MessageCount), + } + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin meta transaction: %w", err) + } + defer tx.Rollback() + for key, value := range values { + if _, err := tx.ExecContext(ctx, "INSERT INTO meta (key, value) VALUES (?, ?)", key, value); err != nil { + return fmt.Errorf("insert meta %q: %w", key, err) + } + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit meta transaction: %w", err) + } + return nil +} + +func selfCheck(ctx context.Context, db *sql.DB, messageCount int, interventions []domain.Intervention) error { + var ftsCount int + if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM messages_fts").Scan(&ftsCount); err != nil { + return fmt.Errorf("count messages_fts: %w", err) + } + if ftsCount != messageCount { + return fmt.Errorf("messages_fts count %d does not match message_count %d", ftsCount, messageCount) + } + if messageCount == 0 { + return nil + } + term := sampleTerm(interventions) + if term == "" { + return nil + } + var matchCount int + if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM messages_fts WHERE messages_fts MATCH ?", term).Scan(&matchCount); err != nil { + return fmt.Errorf("sample FTS match: %w", err) + } + if matchCount == 0 { + return fmt.Errorf("sample FTS match for %q returned no rows", term) + } + return nil +} + +func SHA256File(path string) (string, error) { + file, err := os.Open(path) + if err != nil { + return "", fmt.Errorf("open %s for sha256: %w", path, err) + } + defer file.Close() + hash := sha256.New() + if _, err := io.Copy(hash, file); err != nil { + return "", fmt.Errorf("hash %s: %w", path, err) + } + return hex.EncodeToString(hash.Sum(nil)), nil +} + +func sampleTerm(interventions []domain.Intervention) string { + for _, intervention := range interventions { + for _, message := range intervention.Messages { + for _, field := range strings.Fields(message.Text) { + cleaned := sampleTermPattern.FindString(strings.ToLower(field)) + if len(cleaned) >= 4 { + return cleaned + } + } + } + } + return "" +} + +var sampleTermPattern = regexp.MustCompile(`[[:alnum:]]+`) + +func sittingKey(intervention domain.Intervention) string { + return fmt.Sprintf("%04d-%02d-%s", intervention.ParliamentNumber, intervention.SessionNumber, formatDate(intervention.SittingDate)) +} + +func formatDate(t time.Time) string { + if t.IsZero() { + return "" + } + return t.UTC().Format("2006-01-02") +} + +func speakerName(intervention domain.Intervention) string { + name := strings.TrimSpace(strings.Join([]string{intervention.SpeakerFirstName, intervention.SpeakerLastName}, " ")) + if name == "" { + return "Unknown" + } + return name +} + +func removeSQLiteFiles(path string) error { + for _, suffix := range []string{"", "-wal", "-shm"} { + err := os.Remove(path + suffix) + if err == nil || os.IsNotExist(err) { + continue + } + return fmt.Errorf("remove existing sqlite file %s: %w", path+suffix, err) + } + return nil +} diff --git a/backend/hansard-search-index/internal/adapter/sqlitefts5/builder_test.go b/backend/hansard-search-index/internal/adapter/sqlitefts5/builder_test.go new file mode 100644 index 00000000..e5a0bed8 --- /dev/null +++ b/backend/hansard-search-index/internal/adapter/sqlitefts5/builder_test.go @@ -0,0 +1,130 @@ +package sqlitefts5 + +import ( + "context" + "database/sql" + "os" + "path/filepath" + "testing" + "time" + + "epac/hansard-search-index/internal/adapter/ourcommons" + "epac/hansard-search-index/internal/domain" +) + +func TestBuildCreatesSchemaAndMatchesInsertedMessages(t *testing.T) { + path := filepath.Join(t.TempDir(), "index.sqlite") + builder := NewBuilder(path, fixedClock{}) + _, stats, err := builder.Build(context.Background(), []domain.Intervention{sampleIntervention()}) + if err != nil { + t.Fatalf("Build: %v", err) + } + if stats.MessageCount != 2 { + t.Fatalf("message count = %d, want 2", stats.MessageCount) + } + + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + defer db.Close() + + assertTableExists(t, db, "meta") + assertTableExists(t, db, "interventions") + assertTableExists(t, db, "messages") + assertTableExists(t, db, "messages_fts") + + var count int + if err := db.QueryRow("SELECT COUNT(*) FROM messages_fts WHERE messages_fts MATCH ?", "parliament").Scan(&count); err != nil { + t.Fatalf("match query: %v", err) + } + if count != 1 { + t.Fatalf("match count = %d, want 1", count) + } + + var version string + if err := db.QueryRow("SELECT value FROM meta WHERE key = 'version'").Scan(&version); err != nil { + t.Fatalf("read meta version: %v", err) + } + if version != "v1" { + t.Fatalf("version = %q, want v1", version) + } +} + +func TestFixtureToSQLiteIntegrationMatchesKnownPhrase(t *testing.T) { + fixture := filepath.Join("..", "..", "..", "testdata", "hansard_451_001_slice.xml") + body, err := os.ReadFile(fixture) + if err != nil { + t.Fatalf("read fixture: %v", err) + } + interventions, err := ourcommons.NewParser(nil).Parse(body) + if err != nil { + t.Fatalf("parse fixture: %v", err) + } + + path := filepath.Join(t.TempDir(), "index.sqlite") + _, stats, err := NewBuilder(path, fixedClock{}).Build(context.Background(), interventions) + if err != nil { + t.Fatalf("Build: %v", err) + } + if stats.InterventionCount != 2 || stats.MessageCount != 3 { + t.Fatalf("stats = %#v, want 2 interventions and 3 messages", stats) + } + + db, err := sql.Open("sqlite", path) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + defer db.Close() + + var interventionID string + err = db.QueryRow(` +SELECT i.intervention_id +FROM messages_fts f +JOIN messages m ON m.rowid = f.rowid +JOIN interventions i ON i.rowid = m.intervention_rowid +WHERE messages_fts MATCH ? +ORDER BY m.position +LIMIT 1`, "withdrawing").Scan(&interventionID) + if err != nil { + t.Fatalf("known phrase match: %v", err) + } + if interventionID != "13077673" { + t.Fatalf("match intervention = %q, want 13077673", interventionID) + } +} + +func assertTableExists(t *testing.T, db *sql.DB, table string) { + t.Helper() + var count int + if err := db.QueryRow("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?", table).Scan(&count); err != nil { + t.Fatalf("table lookup %s: %v", table, err) + } + if count != 1 { + t.Fatalf("table %s exists count = %d, want 1", table, count) + } +} + +func sampleIntervention() domain.Intervention { + return domain.Intervention{ + ParliamentNumber: 45, + SessionNumber: 1, + SittingDate: time.Date(2025, 5, 26, 0, 0, 0, 0, time.UTC), + InterventionID: "i-1", + SpeakerFirstName: "John", + SpeakerLastName: "Nater", + PartyAbbreviation: "CPC", + RidingName: "Perth-Wellington", + Topic: "Election of Speaker", + Messages: []domain.Message{ + {MessageID: "m-1", Position: 1, Text: "Parliament should test full text search."}, + {MessageID: "m-2", Position: 2, Text: "The index should keep message rows aligned."}, + }, + } +} + +type fixedClock struct{} + +func (fixedClock) Now() time.Time { + return time.Date(2026, 5, 25, 17, 30, 0, 0, time.UTC) +} diff --git a/backend/hansard-search-index/internal/domain/domain.go b/backend/hansard-search-index/internal/domain/domain.go new file mode 100644 index 00000000..7d421dba --- /dev/null +++ b/backend/hansard-search-index/internal/domain/domain.go @@ -0,0 +1,51 @@ +package domain + +import "time" + +const ManifestVersion = "v1" + +type Session struct { + ParliamentNumber int + SessionNumber int +} + +type Message struct { + MessageID string + Position int + Text string +} + +type Intervention struct { + ParliamentNumber int + SessionNumber int + SittingDate time.Time + InterventionID string + SpeakerFirstName string + SpeakerLastName string + PartyAbbreviation string + RidingName string + Topic string + Messages []Message +} + +type Stats struct { + BuiltAt time.Time + ParliamentNumber int + SessionNumber int + SittingCount int + InterventionCount int + MessageCount int +} + +type Manifest struct { + Version string `json:"version"` + BuiltAt string `json:"built_at"` + ParliamentNumber int `json:"parliament_number"` + SessionNumber int `json:"session_number"` + SittingCount int `json:"sitting_count"` + InterventionCount int `json:"intervention_count"` + MessageCount int `json:"message_count"` + SQLiteKey string `json:"sqlite_key"` + SQLiteSizeBytes int64 `json:"sqlite_size_bytes"` + SQLiteSHA256 string `json:"sqlite_sha256"` +} diff --git a/backend/hansard-search-index/internal/usecase/usecase.go b/backend/hansard-search-index/internal/usecase/usecase.go new file mode 100644 index 00000000..47441cbd --- /dev/null +++ b/backend/hansard-search-index/internal/usecase/usecase.go @@ -0,0 +1,172 @@ +// Package usecase implements the BuildIndex application policy. +// +// It owns the orchestration and port interfaces for fetching current-session +// Hansard XML, parsing source interventions, building a SQLite FTS index, and +// publishing the resulting artifact contract. +package usecase + +import ( + "context" + "errors" + "fmt" + "path" + "strings" + "time" + + "epac/hansard-search-index/internal/domain" +) + +const ( + DefaultPrefix = "hansard-search/v1" + SQLiteName = "index.sqlite" + ManifestName = "manifest.json" +) + +var ( + ErrSittingNotFound = errors.New("hansard sitting not found") + ErrSourceRequired = errors.New("hansard XML source is required") + ErrParserRequired = errors.New("hansard parser is required") + ErrBuilderRequired = errors.New("index builder is required") + ErrUploaderRequired = errors.New("index uploader is required") + ErrManifestRequired = errors.New("manifest writer is required") +) + +type HansardXMLSource interface { + FetchSitting(ctx context.Context, parliament, session, sitting int) ([]byte, error) +} + +type HansardParser interface { + Parse(xml []byte) ([]domain.Intervention, error) +} + +type IndexBuilder interface { + Build(ctx context.Context, interventions []domain.Intervention) (path string, stats domain.Stats, err error) +} + +type IndexUploader interface { + Upload(ctx context.Context, localPath, s3Key string) (sha256 string, sizeBytes int64, err error) +} + +type ManifestWriter interface { + Write(ctx context.Context, manifest domain.Manifest) error +} + +type BuildIndex struct { + source HansardXMLSource + parser HansardParser + builder IndexBuilder + uploader IndexUploader + writer ManifestWriter +} + +type Input struct { + Session domain.Session + Prefix string +} + +type Output struct { + Manifest domain.Manifest +} + +func NewBuildIndex(source HansardXMLSource, parser HansardParser, builder IndexBuilder, uploader IndexUploader, writer ManifestWriter) (*BuildIndex, error) { + switch { + case source == nil: + return nil, ErrSourceRequired + case parser == nil: + return nil, ErrParserRequired + case builder == nil: + return nil, ErrBuilderRequired + case uploader == nil: + return nil, ErrUploaderRequired + case writer == nil: + return nil, ErrManifestRequired + default: + return &BuildIndex{ + source: source, + parser: parser, + builder: builder, + uploader: uploader, + writer: writer, + }, nil + } +} + +func (u *BuildIndex) Execute(ctx context.Context, input Input) (Output, error) { + if input.Session.ParliamentNumber <= 0 { + return Output{}, fmt.Errorf("parliament number must be positive") + } + if input.Session.SessionNumber <= 0 { + return Output{}, fmt.Errorf("session number must be positive") + } + + var all []domain.Intervention + parseFailures := 0 + downloadedSittings := 0 + for sitting := 1; ; sitting++ { + body, err := u.source.FetchSitting(ctx, input.Session.ParliamentNumber, input.Session.SessionNumber, sitting) + if errors.Is(err, ErrSittingNotFound) { + break + } + if err != nil { + return Output{}, fmt.Errorf("fetch sitting %03d: %w", sitting, err) + } + downloadedSittings++ + + interventions, err := u.parser.Parse(body) + if err != nil { + parseFailures++ + continue + } + all = append(all, interventions...) + } + if downloadedSittings > 0 && parseFailures == downloadedSittings { + return Output{}, fmt.Errorf("all %d downloaded sittings failed to parse", downloadedSittings) + } + + localPath, stats, err := u.builder.Build(ctx, all) + if err != nil { + return Output{}, fmt.Errorf("build sqlite index: %w", err) + } + if stats.ParliamentNumber == 0 { + stats.ParliamentNumber = input.Session.ParliamentNumber + } + if stats.SessionNumber == 0 { + stats.SessionNumber = input.Session.SessionNumber + } + if stats.BuiltAt.IsZero() { + stats.BuiltAt = time.Now().UTC() + } + + prefix := cleanPrefix(input.Prefix) + sqliteKey := path.Join(prefix, SQLiteName) + sha256, sizeBytes, err := u.uploader.Upload(ctx, localPath, sqliteKey) + if err != nil { + return Output{}, fmt.Errorf("upload sqlite index: %w", err) + } + + manifest := domain.Manifest{ + Version: domain.ManifestVersion, + BuiltAt: stats.BuiltAt.UTC().Format(time.RFC3339), + ParliamentNumber: stats.ParliamentNumber, + SessionNumber: stats.SessionNumber, + SittingCount: stats.SittingCount, + InterventionCount: stats.InterventionCount, + MessageCount: stats.MessageCount, + SQLiteKey: sqliteKey, + SQLiteSizeBytes: sizeBytes, + SQLiteSHA256: sha256, + } + if err := u.writer.Write(ctx, manifest); err != nil { + return Output{}, fmt.Errorf("write manifest: %w", err) + } + + return Output{Manifest: manifest}, nil +} + +func cleanPrefix(prefix string) string { + prefix = strings.Trim(strings.TrimSpace(prefix), "/") + if prefix == "" || prefix == "." { + return DefaultPrefix + } + return prefix +} diff --git a/backend/hansard-search-index/internal/usecase/usecase_test.go b/backend/hansard-search-index/internal/usecase/usecase_test.go new file mode 100644 index 00000000..e175e522 --- /dev/null +++ b/backend/hansard-search-index/internal/usecase/usecase_test.go @@ -0,0 +1,150 @@ +package usecase + +import ( + "context" + "errors" + "slices" + "testing" + "time" + + "epac/hansard-search-index/internal/domain" +) + +func TestBuildIndexStopsAtFirstMissingSittingAndWritesManifestAfterUpload(t *testing.T) { + clockTick = 0 + ctx := context.Background() + source := &fakeSource{ + bodies: map[int][]byte{ + 1: []byte(""), + 2: []byte(""), + }, + } + parser := &fakeParser{} + builder := &fakeBuilder{stats: domain.Stats{ + BuiltAt: time.Date(2026, 5, 25, 17, 30, 0, 0, time.UTC), + ParliamentNumber: 45, + SessionNumber: 1, + SittingCount: 2, + InterventionCount: 2, + MessageCount: 3, + }} + uploader := &fakeUploader{sha256: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", size: 1234} + writer := &fakeWriter{} + uc, err := NewBuildIndex(source, parser, builder, uploader, writer) + if err != nil { + t.Fatalf("NewBuildIndex: %v", err) + } + + out, err := uc.Execute(ctx, Input{ + Session: domain.Session{ParliamentNumber: 45, SessionNumber: 1}, + Prefix: "hansard-search/v1", + }) + if err != nil { + t.Fatalf("Execute: %v", err) + } + + if !slices.Equal(source.calls, []int{1, 2, 3}) { + t.Fatalf("source calls = %v, want [1 2 3]", source.calls) + } + if uploader.key != "hansard-search/v1/index.sqlite" { + t.Fatalf("uploaded key = %q", uploader.key) + } + if !uploader.calledBefore(writer.calledAt) { + t.Fatal("manifest was written before SQLite upload") + } + if out.Manifest.SQLiteSHA256 != uploader.sha256 { + t.Fatalf("manifest sha = %q", out.Manifest.SQLiteSHA256) + } + if writer.manifest.SQLiteKey != "hansard-search/v1/index.sqlite" { + t.Fatalf("manifest sqlite key = %q", writer.manifest.SQLiteKey) + } + if writer.manifest.MessageCount != 3 { + t.Fatalf("message count = %d, want 3", writer.manifest.MessageCount) + } +} + +func TestBuildIndexFailsWhenEveryDownloadedSittingFailsToParse(t *testing.T) { + source := &fakeSource{bodies: map[int][]byte{1: []byte("")}} + parser := &fakeParser{err: errors.New("malformed xml")} + builder := &fakeBuilder{} + uploader := &fakeUploader{} + writer := &fakeWriter{} + uc, err := NewBuildIndex(source, parser, builder, uploader, writer) + if err != nil { + t.Fatalf("NewBuildIndex: %v", err) + } + + _, err = uc.Execute(context.Background(), Input{ + Session: domain.Session{ParliamentNumber: 45, SessionNumber: 1}, + }) + if err == nil { + t.Fatal("expected parse failure, got nil") + } +} + +type fakeSource struct { + bodies map[int][]byte + calls []int +} + +func (s *fakeSource) FetchSitting(_ context.Context, _, _, sitting int) ([]byte, error) { + s.calls = append(s.calls, sitting) + if body, ok := s.bodies[sitting]; ok { + return body, nil + } + return nil, ErrSittingNotFound +} + +type fakeParser struct { + err error +} + +func (p *fakeParser) Parse(body []byte) ([]domain.Intervention, error) { + if p.err != nil { + return nil, p.err + } + return []domain.Intervention{{ + InterventionID: string(body), + Messages: []domain.Message{{MessageID: string(body) + "-p1", Position: 1, Text: "sample parliament text"}}, + }}, nil +} + +type fakeBuilder struct { + stats domain.Stats +} + +func (b *fakeBuilder) Build(context.Context, []domain.Intervention) (string, domain.Stats, error) { + return "/tmp/index.sqlite", b.stats, nil +} + +type fakeUploader struct { + key string + sha256 string + size int64 + calledTick int +} + +var clockTick int + +func (u *fakeUploader) Upload(_ context.Context, _, key string) (string, int64, error) { + clockTick++ + u.calledTick = clockTick + u.key = key + return u.sha256, u.size, nil +} + +func (u *fakeUploader) calledBefore(tick int) bool { + return u.calledTick > 0 && tick > u.calledTick +} + +type fakeWriter struct { + manifest domain.Manifest + calledAt int +} + +func (w *fakeWriter) Write(_ context.Context, manifest domain.Manifest) error { + clockTick++ + w.calledAt = clockTick + w.manifest = manifest + return nil +} diff --git a/backend/hansard-search-index/main.go b/backend/hansard-search-index/main.go new file mode 100644 index 00000000..904dd017 --- /dev/null +++ b/backend/hansard-search-index/main.go @@ -0,0 +1,126 @@ +// hansard-search-index Lambda builds the SQLite FTS5 Hansard search artifact. +package main + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "strconv" + "strings" + "time" + + ourcommonsadapter "epac/hansard-search-index/internal/adapter/ourcommons" + s3adapter "epac/hansard-search-index/internal/adapter/s3" + "epac/hansard-search-index/internal/adapter/sqlitefts5" + "epac/hansard-search-index/internal/domain" + "epac/hansard-search-index/internal/usecase" + "epac/observability" + + "github.com/aws/aws-lambda-go/lambda" + "github.com/aws/aws-sdk-go-v2/config" + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" +) + +const ( + DefaultParliamentNumber = 45 + DefaultSessionNumber = 1 +) + +func HandleRequest(ctx context.Context) error { + session, err := sessionFromEnv() + if err != nil { + return err + } + bucket := strings.TrimSpace(os.Getenv("EPAC_ARTIFACT_BUCKET")) + if bucket == "" { + return fmt.Errorf("EPAC_ARTIFACT_BUCKET is required") + } + prefix := firstEnv("EPAC_HANSARD_SEARCH_PREFIX") + if prefix == "" { + prefix = usecase.DefaultPrefix + } + + logger := ourcommonsadapter.NewJSONLogger(os.Stdout) + source := ourcommonsadapter.NewSource( + ourcommonsadapter.WithHTTPClient(&http.Client{Timeout: 30 * time.Second}), + ourcommonsadapter.WithLogger(logger), + ) + parser := ourcommonsadapter.NewParser(logger) + builder := sqlitefts5.NewBuilder(sqlitefts5.DefaultPath, sqlitefts5.SystemClock{}) + + awsCfg, err := config.LoadDefaultConfig(ctx) + if err != nil { + return fmt.Errorf("load AWS config: %w", err) + } + store := s3adapter.NewStore(awss3.NewFromConfig(awsCfg), bucket, prefix) + + buildIndex, err := usecase.NewBuildIndex(source, parser, builder, store, store) + if err != nil { + return err + } + out, err := buildIndex.Execute(ctx, usecase.Input{Session: session, Prefix: prefix}) + if err != nil { + return err + } + logSummary(out.Manifest) + return nil +} + +func sessionFromEnv() (domain.Session, error) { + parliament, err := positiveIntFromEnv("PARLIAMENT_NUMBER", DefaultParliamentNumber) + if err != nil { + return domain.Session{}, err + } + session, err := positiveIntFromEnv("SESSION_NUMBER", DefaultSessionNumber) + if err != nil { + return domain.Session{}, err + } + return domain.Session{ParliamentNumber: parliament, SessionNumber: session}, nil +} + +func positiveIntFromEnv(name string, fallback int) (int, error) { + raw := strings.TrimSpace(os.Getenv(name)) + if raw == "" { + return fallback, nil + } + value, err := strconv.Atoi(raw) + if err != nil || value <= 0 { + return 0, fmt.Errorf("%s must be a positive integer", name) + } + return value, nil +} + +func firstEnv(names ...string) string { + for _, name := range names { + if value := strings.TrimSpace(os.Getenv(name)); value != "" { + return value + } + } + return "" +} + +func logSummary(manifest domain.Manifest) { + record := map[string]any{ + "timestamp": time.Now().UTC().Format(time.RFC3339), + "level": "info", + "pipeline": "hansard-search-index", + "event": "index_built", + "sqlite_key": manifest.SQLiteKey, + "sqlite_size_bytes": manifest.SQLiteSizeBytes, + "sqlite_sha256": manifest.SQLiteSHA256, + "sitting_count": manifest.SittingCount, + "intervention_count": manifest.InterventionCount, + "message_count": manifest.MessageCount, + } + data, err := json.Marshal(record) + if err != nil { + return + } + fmt.Println(string(data)) +} + +func main() { + lambda.Start(observability.WrapNoEvent("hansard-search-index", HandleRequest)) +} diff --git a/backend/hansard-search-index/main_test.go b/backend/hansard-search-index/main_test.go new file mode 100644 index 00000000..e714a69f --- /dev/null +++ b/backend/hansard-search-index/main_test.go @@ -0,0 +1,39 @@ +package main + +import "testing" + +func TestSessionFromEnvUsesDefaultsAndAllowsOverrides(t *testing.T) { + t.Setenv("PARLIAMENT_NUMBER", "") + t.Setenv("SESSION_NUMBER", "") + session, err := sessionFromEnv() + if err != nil { + t.Fatalf("sessionFromEnv default: %v", err) + } + if session.ParliamentNumber != DefaultParliamentNumber || session.SessionNumber != DefaultSessionNumber { + t.Fatalf("default session = %#v", session) + } + + t.Setenv("PARLIAMENT_NUMBER", "44") + t.Setenv("SESSION_NUMBER", "1") + session, err = sessionFromEnv() + if err != nil { + t.Fatalf("sessionFromEnv override: %v", err) + } + if session.ParliamentNumber != 44 || session.SessionNumber != 1 { + t.Fatalf("override session = %#v", session) + } +} + +func TestSessionFromEnvRejectsInvalidOverride(t *testing.T) { + t.Setenv("PARLIAMENT_NUMBER", "nope") + t.Setenv("SESSION_NUMBER", "1") + if _, err := sessionFromEnv(); err == nil { + t.Fatal("expected invalid parliament error") + } + + t.Setenv("PARLIAMENT_NUMBER", "45") + t.Setenv("SESSION_NUMBER", "0") + if _, err := sessionFromEnv(); err == nil { + t.Fatal("expected invalid session error") + } +} diff --git a/backend/hansard-search-index/testdata/hansard_451_001_slice.xml b/backend/hansard-search-index/testdata/hansard_451_001_slice.xml new file mode 100644 index 00000000..bff6e3bb --- /dev/null +++ b/backend/hansard-search-index/testdata/hansard_451_001_slice.xml @@ -0,0 +1,37 @@ + + + + Monday, May 26, 2025 + 1 + 45 + + + + First Session—45th Parliament + + (0845) + [Translation] + Election of Speaker + Election of Speaker + + + + John Nater (Perth—Wellington, CPC): + + Mr. Chair, dean of the House of Commons, I am very pleased to see you presiding over the election of a Speaker for the seventh time, and I wish to inform you that I am withdrawing my name from the list of candidates. + + + [English] + + + Chris d'Entremont (Acadie—Annapolis, CPC): + + A good Speaker knows when to keep their speeches short. + Mr. Chair, I would ask you to withdraw my name from the ballot, please. + + + + + + + diff --git a/backend/manifest/deployment-services.json b/backend/manifest/deployment-services.json index 19aece99..46c71843 100644 --- a/backend/manifest/deployment-services.json +++ b/backend/manifest/deployment-services.json @@ -67,6 +67,14 @@ "production": {"database": false, "artifact": true} } }, + { + "name": "hansard-search-index", + "deploy": {"staging": true, "production": true}, + "sync": { + "staging": {"database": false, "artifact": true}, + "production": {"database": false, "artifact": true} + } + }, { "name": "member-speeches", "deploy": {"staging": true, "production": true}, diff --git a/docs/architecture/use-case-catalog.md b/docs/architecture/use-case-catalog.md index 4cc3e5ee..c31d9f5a 100644 --- a/docs/architecture/use-case-catalog.md +++ b/docs/architecture/use-case-catalog.md @@ -35,6 +35,8 @@ For the Clean Architecture shape this catalog assumes, see [`docs/architecture/` | `ArtifactKey` | A published object key used by backend artifact publishers, such as `members/v1/all.json`. | | `Manifest` | The root manifest.json document: schema version, generation timestamp, and sorted artifact entries. | | `ManifestEntry` | Metadata for one S3 artifact: key, size, SHA-256 hash, ETag, last-modified, and per-artifact schema version. | +| `HansardSearchIntervention` | Backend-only parsed Hansard intervention value used while building the SQLite FTS5 search artifact. | +| `HansardSearchManifest` | Backend-only manifest pointer for the current-session Hansard search SQLite artifact. | ## Ports @@ -57,6 +59,10 @@ For the Clean Architecture shape this catalog assumes, see [`docs/architecture/` | `RidingBoundaryRepository` | outbound | Read federal riding boundary artifacts by slug. | | `CalendarArtifactRepository` | outbound | Read the published House sitting calendar ICS artifact. | | `AppConfigRepository` | outbound | Read backend-provided app configuration artifacts. | +| `HansardXMLSource` | outbound | Fetch authoritative ourcommons.ca Hansard XML by parliament, session, and sitting number. | +| `HansardParser` | outbound | Parse Hansard XML into intervention and paragraph values without coupling the use case to `encoding/xml`. | +| `HansardSearchIndexBuilder` | outbound | Build and self-check a local SQLite FTS5 index from parsed interventions. | +| `HansardSearchArtifactStore` | outbound | Upload the SQLite search index and write the manifest pointer to S3. | --- @@ -170,6 +176,30 @@ Current implementation: --- +### BuildIndex + +``` +Actor: Operator (manual GitHub Actions dispatch or aws lambda invoke) +Goal: Build the current-session Hansard SQLite FTS5 index and publish its manifest pointer. +Inputs: Parliament number, session number, artifact bucket, artifact prefix. +Outputs: index.sqlite and manifest.json under the configured S3 prefix. +Entities / values: HansardSearchIntervention, HansardSearchManifest, ArtifactKey. +Ports: HansardXMLSource, HansardParser, HansardSearchIndexBuilder, HansardSearchArtifactStore. +Primary adapters: ourcommons.ca XML source/parser, modernc.org/sqlite FTS5 builder, AWS S3 artifact writer. +Current implementation: + backend/hansard-search-index/main.go + backend/hansard-search-index/internal/usecase/usecase.go + backend/hansard-search-index/internal/domain/domain.go + backend/hansard-search-index/internal/adapter/ourcommons/source.go + backend/hansard-search-index/internal/adapter/ourcommons/parser.go + backend/hansard-search-index/internal/adapter/sqlitefts5/builder.go + backend/hansard-search-index/internal/adapter/s3/s3.go +``` + +> **Boundary rule:** `backend/hansard-search-index/internal/usecase/` owns the application policy and imports only standard library packages plus the local domain package. HTTP, XML decoding, SQLite, AWS SDK, and Lambda runtime wiring stay in adapters or `main.go`. + +--- + ### FindMyMP ``` diff --git a/infra/terraform/core/iam.tf b/infra/terraform/core/iam.tf index 6cfd435c..28f9f7ce 100644 --- a/infra/terraform/core/iam.tf +++ b/infra/terraform/core/iam.tf @@ -1,5 +1,6 @@ locals { - account_id = "227530433709" + account_id = "227530433709" + hansard_search_prefix = trim(var.hansard_search_prefix, "/") backend_ci_roles = { staging = "GitHubActions-epac-backend-staging" @@ -174,3 +175,27 @@ resource "aws_iam_role_policy" "backend_production_ci" { role = local.backend_ci_roles.production policy = data.aws_iam_policy_document.backend_production_ci.json } + +data "aws_iam_policy_document" "lambda_hansard_search_index_artifacts" { + statement { + sid = "AllowHansardSearchIndexArtifactWrites" + effect = "Allow" + + actions = [ + # AWS authorizes HeadObject through s3:GetObject; there is no separate + # s3:HeadObject IAM action. + "s3:GetObject", + "s3:PutObject", + ] + + resources = [ + "arn:aws:s3:::${var.artifacts_bucket_name}/${local.hansard_search_prefix}/*", + ] + } +} + +resource "aws_iam_role_policy" "lambda_hansard_search_index_artifacts" { + name = "epac-hansard-search-index-artifacts" + role = var.lambda_role_name + policy = data.aws_iam_policy_document.lambda_hansard_search_index_artifacts.json +} diff --git a/infra/terraform/core/variables.tf b/infra/terraform/core/variables.tf index df28eccc..0b41b2b0 100644 --- a/infra/terraform/core/variables.tf +++ b/infra/terraform/core/variables.tf @@ -21,3 +21,15 @@ variable "artifacts_route53_zone_name" { type = string default = "riddimsoftware.com" } + +variable "lambda_role_name" { + description = "Existing EPAC Lambda execution role name." + type = string + default = "epac-lambda-role" +} + +variable "hansard_search_prefix" { + description = "S3 key prefix used by the Hansard search index artifact." + type = string + default = "hansard-search/v1" +} diff --git a/infra/terraform/production/main.tf b/infra/terraform/production/main.tf index 3abf2d8d..61519f63 100644 --- a/infra/terraform/production/main.tf +++ b/infra/terraform/production/main.tf @@ -36,6 +36,17 @@ locals { for service in keys(local.http_services) : service => "apigw-epac-api-${service}" } + + canonical_function_services = toset([ + "hansard-search-index", + ]) + + lambda_config = { + "hansard-search-index" = { + timeout = 900 + memory_size = 1024 + } + } } # Production HTTP API (existing API ID smun5g2szc; import before apply). @@ -78,12 +89,14 @@ resource "aws_apigatewayv2_stage" "production" { resource "aws_lambda_function" "production" { for_each = toset(local.services) - function_name = each.key + function_name = contains(local.canonical_function_services, each.key) ? "epac-${each.key}-production" : each.key role = var.lambda_role_arn runtime = "provided.al2023" architectures = ["arm64"] handler = "bootstrap" publish = false + timeout = try(local.lambda_config[each.key].timeout, null) + memory_size = try(local.lambda_config[each.key].memory_size, null) filename = "${path.module}/placeholder.zip" diff --git a/infra/terraform/staging/main.tf b/infra/terraform/staging/main.tf index 37bbf4ee..56042a9c 100644 --- a/infra/terraform/staging/main.tf +++ b/infra/terraform/staging/main.tf @@ -29,6 +29,13 @@ locals { for route in local.staging_api_routes : "${route.service}::${route.route_key}" => route } + + lambda_config = { + "hansard-search-index" = { + timeout = 900 + memory_size = 1024 + } + } } # Lambda functions — code is managed by the staging deploy workflow, not Terraform. @@ -42,6 +49,8 @@ resource "aws_lambda_function" "staging" { architectures = ["arm64"] handler = "bootstrap" publish = false + timeout = try(local.lambda_config[each.key].timeout, null) + memory_size = try(local.lambda_config[each.key].memory_size, null) # Placeholder zip — CI overwrites code on every staging deploy. filename = "${path.module}/placeholder.zip" diff --git a/scripts/ci/backend_staging_smoke.py b/scripts/ci/backend_staging_smoke.py index 5095849b..f8c46cfc 100755 --- a/scripts/ci/backend_staging_smoke.py +++ b/scripts/ci/backend_staging_smoke.py @@ -12,12 +12,15 @@ import argparse import json import os +import re +import subprocess import sys import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass +from tempfile import NamedTemporaryFile from typing import Any, Callable @@ -44,6 +47,7 @@ class SmokeCheck: fixture_note: str = "" # When False, the validator receives raw response bytes instead of parsed JSON. expect_json: bool = True + kind: str = "http" def url(self, base_url: str) -> str: base = base_url.rstrip("/") @@ -179,6 +183,23 @@ def validate_openapi(status: int, payload: Any) -> None: raise SmokeFailure("openapi-json: response body missing 'paths' key") +def validate_hansard_search_manifest(status: int, payload: Any) -> None: + if status == 404: + return + body = require_dict(payload, "hansard-search-manifest") + require_keys( + body, + "hansard-search-manifest", + {"version", "sitting_count", "sqlite_sha256"}, + ) + if body["version"] != "v1": + raise SmokeFailure(f"hansard-search-manifest: version = {body['version']!r}, want 'v1'") + if not isinstance(body["sitting_count"], int) or body["sitting_count"] < 0: + raise SmokeFailure("hansard-search-manifest: sitting_count must be a non-negative integer") + if not isinstance(body["sqlite_sha256"], str) or not re.fullmatch(r"[a-f0-9]{64}", body["sqlite_sha256"]): + raise SmokeFailure("hansard-search-manifest: sqlite_sha256 must be a 64-character lowercase hex string") + + CHECKS = [ # --- health --- SmokeCheck( @@ -191,6 +212,18 @@ def validate_openapi(status: int, payload: Any) -> None: deterministic_note="Contract check accepts ok/degraded HealthResponse and catches DB/Lambda error bodies.", fixture_note="Pipeline freshness can make this degraded until staging data jobs are seeded and scheduled.", ), + # --- hansard search index manifest --- + SmokeCheck( + name="hansard-search-manifest", + method="S3", + path="hansard-search/v1/manifest.json", + query={}, + expected_statuses={200, 404}, + validator=validate_hansard_search_manifest, + deterministic_note="Contract check validates the v1 manifest envelope and SHA-256 format when the index has been generated.", + fixture_note="The first deploy may not have a manifest until an operator runs the manual reindex; HTTP 404 is reported as a skip warning.", + kind="s3-hansard-search-manifest", + ), # --- member speeches --- SmokeCheck( name="member-speeches:min-args", @@ -374,6 +407,9 @@ def validate_openapi(status: int, payload: Any) -> None: def fetch_response(check: SmokeCheck, base_url: str) -> tuple[int, Any]: + if check.kind == "s3-hansard-search-manifest": + return fetch_hansard_search_manifest() + headers = { "Accept": "application/json", "User-Agent": "epac-staging-smoke/1.0", @@ -408,6 +444,60 @@ def fetch_response(check: SmokeCheck, base_url: str) -> tuple[int, Any]: return status, payload +def fetch_hansard_search_manifest() -> tuple[int, Any]: + bucket = first_env( + "EPAC_ARTIFACT_BUCKET_STAGING", + "EPAC_ARTIFACT_BUCKET", + "ARTIFACTS_BUCKET", + "ARTIFACT_BUCKET", + ) + prefix = os.environ.get("EPAC_HANSARD_SEARCH_PREFIX", "hansard-search/v1").strip().strip("/") + key = f"{prefix}/manifest.json" + if not bucket: + return 404, {"_smoke_evidence": "SKIP artifact bucket not configured"} + + with NamedTemporaryFile(delete=False) as handle: + output_path = handle.name + try: + result = subprocess.run( + [ + "aws", + "s3api", + "get-object", + "--bucket", + bucket, + "--key", + key, + output_path, + ], + check=False, + capture_output=True, + text=True, + ) + if result.returncode != 0: + stderr = result.stderr or result.stdout + if any(fragment in stderr for fragment in ("NoSuchKey", "Not Found", "404", "NoSuchBucket")): + return 404, {"_smoke_evidence": f"SKIP s3://{bucket}/{key} not found"} + raise SmokeFailure(f"hansard-search-manifest: aws s3api get-object failed: {stderr.strip()}") + with open(output_path, "r", encoding="utf-8") as manifest_file: + return 200, json.load(manifest_file) + except FileNotFoundError as error: + raise SmokeFailure("hansard-search-manifest: aws CLI is not installed") from error + finally: + try: + os.unlink(output_path) + except OSError: + pass + + +def first_env(*names: str) -> str: + for name in names: + value = os.environ.get(name, "").strip() + if value: + return value + return "" + + def run_check(check: SmokeCheck, base_url: str) -> tuple[bool, str]: last_error = "" for attempt in range(1, RETRIES + 1): @@ -417,6 +507,8 @@ def run_check(check: SmokeCheck, base_url: str) -> tuple[bool, str]: expected = ", ".join(str(code) for code in sorted(check.expected_statuses)) raise SmokeFailure(f"{check.name}: expected HTTP {expected}, got {status}") check.validator(status, payload) + if isinstance(payload, dict) and "_smoke_evidence" in payload: + return True, str(payload["_smoke_evidence"]) return True, f"HTTP {status}" except SmokeFailure as error: last_error = str(error)