diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4e7c9dbf9..0c9d711bc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -81,8 +81,10 @@ jobs: go-version-file: go.mod cache: ${{ github.repository == 'kenn-io/msgvault' && (github.event_name == 'push' && github.ref == 'refs/heads/main' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name == github.repository && github.event.pull_request.base.repo.full_name == github.repository)) && 'false' || 'true' }} - - name: Install golangci-lint - run: go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@v2.12.2 + - name: Install lint tools + run: | + make lint-tools + echo "$(go env GOPATH)/bin" >> "$GITHUB_PATH" - name: Test run: make test @@ -93,11 +95,8 @@ jobs: - name: Lint run: make lint-ci - - name: Install govulncheck - run: go install golang.org/x/vuln/cmd/govulncheck@v1.1.4 - - name: Vulnerability check - run: govulncheck -tags "fts5 sqlite_vec" ./... + run: make vulncheck test-macos-15: if: github.repository == 'kenn-io/msgvault' && github.event_name == 'push' && github.ref == 'refs/heads/main' diff --git a/.golangci.yml b/.golangci.yml index 0ec0fc164..69b028d21 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -1,6 +1,7 @@ version: "2" run: + go: "1.27" # Lint the build-tagged code, not just the untagged subset. The primary # build uses `fts5 sqlite_vec` (Makefile BUILD_TAGS) and the pgvector # backend is gated behind `pgvector`; without these tags golangci-lint @@ -47,7 +48,7 @@ linters: - godot - goheader - gomoddirectives - - gomodguard + - gomodguard_v2 - goprintffuncname - gosec - govet diff --git a/Dockerfile b/Dockerfile index bdac74120..a962e4778 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,7 +12,7 @@ RUN cd web && bun run generate && bun run build # Go build stage. # Pin by digest for reproducibility; update periodically. -FROM golang:1.26.6-bookworm@sha256:116d58cbd88c1297624acc6e967a060012422bacf9930927e23fb719189c6f36 AS builder +FROM golang:1.27.0-bookworm@sha256:484ef6066fa69acb059fdfeda7ba2b8f7391f2ef6abc6f9b8411e669ebd56466 AS builder # Install build dependencies for CGO (SQLite, DuckDB). # libsqlite3-dev provides sqlite3.h, required to compile the sqlite-vec diff --git a/Makefile b/Makefile index ab361fbe5..08ab87361 100644 --- a/Makefile +++ b/Makefile @@ -17,6 +17,15 @@ LDFLAGS_RELEASE := $(LDFLAGS) -s -w # - sqlite_vec: enable the sqlite-vec extension for vector search BUILD_TAGS := fts5 sqlite_vec TEST_TIMEOUT := 60m +GOLANGCI_LINT_VERSION ?= v2.13.1 +GOVULNCHECK_VERSION ?= v1.7.0 +GO_INSTALL_BIN := $(shell go env GOBIN) +ifeq ($(strip $(GO_INSTALL_BIN)),) +GO_INSTALL_BIN := $(shell go env GOPATH)/bin +endif +GOLANGCI_LINT_BIN := $(GO_INSTALL_BIN)/golangci-lint +CI_TOOLS_BIN := $(shell git rev-parse --path-format=absolute --git-path ci-tools/bin) +GOVULNCHECK_BIN := $(CI_TOOLS_BIN)/govulncheck # Build tags for the PostgreSQL test lane (test-pg). Must be the full build set: # pgvector gates the vector-on-PG code paths (//go:build pgvector), and sqlite_vec @@ -53,7 +62,7 @@ export GOLANGCI_LINT_CACHE # serialize one another while duplicate runners in one worktree can wait. GOLANGCI_LINT_TMP ?= $(GOLANGCI_LINT_CACHE)/tmp -.PHONY: build build-release install clean test test-v test-pg test-pg-shipped test-pg-both pg-shipped-only-check require-test-db fmt lint lint-ci testify-helper-check tidy openapi api-generate openapi-check api-check web-install web-generate web-check web-test web-test-browser web-e2e web-build web-embed web-assets-check smoke-web-release shootout run-shootout install-hooks bench vcard-registry-check vcard-registry-update docs-install docs-build docs-serve docs-check docs-fixture-test docs-fixture-check docs-fixture-smoke docs-web-screenshots docs-screenshots docs-assets-branch docs-generated-assets-branch docs-deploy-staging docs-deploy help +.PHONY: build build-release install clean test test-v test-pg test-pg-shipped test-pg-both pg-shipped-only-check require-test-db fmt lint-tools lint lint-ci vuln-tools vulncheck testify-helper-check tidy openapi api-generate openapi-check api-check web-install web-generate web-check web-test web-test-browser web-e2e web-build web-embed web-assets-check smoke-web-release shootout run-shootout install-hooks bench vcard-registry-check vcard-registry-update docs-install docs-build docs-serve docs-check docs-fixture-test docs-fixture-check docs-fixture-smoke docs-web-screenshots docs-screenshots docs-assets-branch docs-generated-assets-branch docs-deploy-staging docs-deploy help # Build the binary (debug) build: web-embed @@ -272,6 +281,10 @@ smoke-web-release: fmt: go fmt ./... +# Install the pinned linter used by CI. +lint-tools: + go install github.com/golangci/golangci-lint/v2/cmd/golangci-lint@$(GOLANGCI_LINT_VERSION) + # Run linter (auto-fix) lint: @if ! command -v golangci-lint >/dev/null 2>&1; then \ @@ -282,13 +295,22 @@ lint: TMPDIR="$(GOLANGCI_LINT_TMP)" golangci-lint run --fix ./... # Run linter (CI, no auto-fix) -lint-ci: testify-helper-check - @if ! command -v golangci-lint >/dev/null 2>&1; then \ - echo "golangci-lint not found. Install: https://golangci-lint.run/usage/install/" >&2; \ - exit 1; \ - fi +lint-ci: lint-tools testify-helper-check @mkdir -p "$(GOLANGCI_LINT_TMP)" - TMPDIR="$(GOLANGCI_LINT_TMP)" golangci-lint run ./... + TMPDIR="$(GOLANGCI_LINT_TMP)" "$(GOLANGCI_LINT_BIN)" run ./... + @if [ -n "$$GITHUB_PATH" ]; then \ + $(MAKE) --no-print-directory vuln-tools; \ + printf '%s\n' "$(CI_TOOLS_BIN)" >> "$$GITHUB_PATH"; \ + fi + +# Install and run the scanner from a repository-owned path so a stale tool +# installed by the base branch's pull-request workflow cannot replace it. +vuln-tools: + @mkdir -p "$(CI_TOOLS_BIN)" + GOBIN="$(CI_TOOLS_BIN)" go install golang.org/x/vuln/cmd/govulncheck@$(GOVULNCHECK_VERSION) + +vulncheck: vuln-tools + "$(GOVULNCHECK_BIN)" -tags "$(BUILD_TAGS)" ./... # Enforce testify helper usage in assertion-heavy tests testify-helper-check: @@ -398,6 +420,7 @@ help: @echo " fmt - Format code" @echo " lint - Run linter (auto-fix)" @echo " lint-ci - Run linter (CI, no auto-fix; also runs testify-helper-check)" + @echo " vulncheck - Run the pinned Go vulnerability scanner" @echo " testify-helper-check - Enforce testify helper usage in assertion-heavy tests" @echo " tidy - Tidy go.mod" @echo " vcard-registry-check - Network-check IANA registry drift (manual; not CI)" diff --git a/README.md b/README.md index 835720744..0414ce91e 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@

msgvault

- Go 1.26+ + Go 1.27+ License: MIT Docs Discord @@ -73,7 +73,7 @@ powershell -ExecutionPolicy ByPass -c "irm https://msgvault.io/install.ps1 | iex The installer detects your OS and architecture, downloads the latest release from [GitHub Releases](https://github.com/kenn-io/msgvault/releases), verifies the SHA-256 checksum, and installs the binary. You can review the script ([bash](https://msgvault.io/install.sh), [PowerShell](https://msgvault.io/install.ps1)) before running, or download a release binary directly from GitHub. -To build from source instead (requires **Go 1.26+**, **Bun 1.3.14+**, and a +To build from source instead (requires **Go 1.27+**, **Bun 1.3.14+**, and a C/C++ compiler for CGO and to statically link DuckDB): ```bash diff --git a/cmd/msgvault/cmd/add_discord.go b/cmd/msgvault/cmd/add_discord.go index 29ed94fe1..8f20bfafb 100644 --- a/cmd/msgvault/cmd/add_discord.go +++ b/cmd/msgvault/cmd/add_discord.go @@ -259,8 +259,7 @@ func messageContentUnavailable(message discord.Message) bool { } func discordDiagnostic(err error) string { - var apiErr *discord.APIError - if errors.As(err, &apiErr) { + if apiErr, ok := errors.AsType[*discord.APIError](err); ok { switch apiErr.StatusCode { case http.StatusUnauthorized: return "authentication failed" diff --git a/cmd/msgvault/cmd/addaccount.go b/cmd/msgvault/cmd/addaccount.go index 36927f4d8..35ba087e7 100644 --- a/cmd/msgvault/cmd/addaccount.go +++ b/cmd/msgvault/cmd/addaccount.go @@ -435,8 +435,7 @@ func runAddAccountLocal(cmd *cobra.Command, args []string) error { return fmt.Errorf("service account token for %s: %w", email, saErr) } if saErr := oauth.ValidateTokenEmail(cmd.Context(), ts, email); saErr != nil { - var mismatch *oauth.TokenMismatchError - if errors.As(saErr, &mismatch) { + if mismatch, ok := errors.AsType[*oauth.TokenMismatchError](saErr); ok { existing, lookupErr := findGmailSource(s, email) if lookupErr != nil && !errors.Is(lookupErr, errGmailSourceNotFound) { return fmt.Errorf("service account validation failed: %w (also: %w)", saErr, lookupErr) diff --git a/cmd/msgvault/cmd/repair_dates.go b/cmd/msgvault/cmd/repair_dates.go index 52c01f17a..dab5001ac 100644 --- a/cmd/msgvault/cmd/repair_dates.go +++ b/cmd/msgvault/cmd/repair_dates.go @@ -322,13 +322,11 @@ func scanAndPlanDateRepairs( continue } plan.repairs = append(plan.repairs, plannedDateRepair{ - MessageDateRepair: store.MessageDateRepair{ - ID: candidate.ID, - SentAt: resolved, - ExpectedLastModifiedAt: candidate.LastModifiedAt, - }, - source: source, - oldSentAt: candidate.SentAt, + ID: candidate.ID, + SentAt: resolved, + ExpectedLastModifiedAt: candidate.LastModifiedAt, + source: source, + oldSentAt: candidate.SentAt, }) } return plan, nil diff --git a/cmd/msgvault/cmd/repair_dates_test.go b/cmd/msgvault/cmd/repair_dates_test.go index 6f730e7b6..c6f089add 100644 --- a/cmd/msgvault/cmd/repair_dates_test.go +++ b/cmd/msgvault/cmd/repair_dates_test.go @@ -328,10 +328,8 @@ func TestApplyPlannedDateRepairsRecordsGuardFailure(t *testing.T) { plan := &dateRepairPlan{ candidateCount: 1, repairs: []plannedDateRepair{{ - MessageDateRepair: store.MessageDateRepair{ - ID: messageID, - SentAt: time.Date(2007, 1, 2, 15, 4, 5, 0, time.UTC), - }, + ID: messageID, + SentAt: time.Date(2007, 1, 2, 15, 4, 5, 0, time.UTC), source: mime.DateSourceReceived, oldSentAt: sql.NullTime{ Time: time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC), diff --git a/cmd/msgvault/cmd/root.go b/cmd/msgvault/cmd/root.go index d88d684a5..b57ca7d6c 100644 --- a/cmd/msgvault/cmd/root.go +++ b/cmd/msgvault/cmd/root.go @@ -444,8 +444,7 @@ func wrapOAuthError(err error) error { // permanently invalid (expired or revoked), as opposed to a transient failure // like a network error or context cancellation. func isAuthInvalidError(err error) bool { - var retrieveErr *oauth2.RetrieveError - if errors.As(err, &retrieveErr) { + if retrieveErr, ok := errors.AsType[*oauth2.RetrieveError](err); ok { // Google returns "invalid_grant" when refresh tokens are expired or revoked return retrieveErr.ErrorCode == "invalid_grant" } @@ -586,8 +585,7 @@ func getTokenSourceWithReauth( // AuthorizeManual validates the token and atomically saves it, // so the old token is only overwritten after validation succeeds. if authErr := authorizeManualForReauth(ctx, mgr, email); authErr != nil { - var mismatch *oauth.TokenMismatchError - if errors.As(authErr, &mismatch) { + if mismatch, ok := errors.AsType[*oauth.TokenMismatchError](authErr); ok { return nil, fmt.Errorf( "re-authorize %s: %w\n"+ "If this account uses an alias, remove "+ diff --git a/cmd/msgvault/cmd/search.go b/cmd/msgvault/cmd/search.go index ed31a7683..0ff09248d 100644 --- a/cmd/msgvault/cmd/search.go +++ b/cmd/msgvault/cmd/search.go @@ -221,8 +221,6 @@ func runHTTPSearch(cmd *cobra.Command, queryStr string) error { // nil error return mirrors outputSearchResultsJSON so callers can return // either uniformly; tabwriter output never fails. -// -//nolint:unparam // symmetry with error-returning outputSearchResultsJSON sibling func outputSearchResultsTable(results []query.MessageSummary) error { w := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0) _, _ = fmt.Fprintln(w, "ID\tDATE\tFROM\tSUBJECT\tSIZE") diff --git a/cmd/msgvault/cmd/show_message.go b/cmd/msgvault/cmd/show_message.go index e176bedb7..3a01e52eb 100644 --- a/cmd/msgvault/cmd/show_message.go +++ b/cmd/msgvault/cmd/show_message.go @@ -88,8 +88,6 @@ func showHTTPMessage(cmd *cobra.Command, idStr string) error { // nil error return mirrors outputMessageJSON so callers can return either // uniformly; text printing never fails. -// -//nolint:unparam // symmetry with error-returning outputMessageJSON sibling func outputMessageText(msg *query.MessageDetail) error { // Header section fmt.Println("═══════════════════════════════════════════════════════════════════════════════") diff --git a/docs/development.md b/docs/development.md index 67abd954f..35e4254e9 100644 --- a/docs/development.md +++ b/docs/development.md @@ -5,7 +5,7 @@ description: Build, test, lint, and code conventions. ## Build -Source builds require Go 1.26+, Bun 1.3.14+, and a C/C++ compiler. The Make +Source builds require Go 1.27+, Bun 1.3.14+, and a C/C++ compiler. The Make targets install the pinned browser dependencies when `web/package.json` or `web/bun.lock` changes, then embed the production UI in the Go binary. diff --git a/docs/screenshots/Dockerfile b/docs/screenshots/Dockerfile index 709438f5c..e5c9c6c17 100644 --- a/docs/screenshots/Dockerfile +++ b/docs/screenshots/Dockerfile @@ -1,6 +1,6 @@ # syntax=docker/dockerfile:1.4 # Stage 1: Build msgvault binary -FROM golang:1.26-bookworm AS builder +FROM golang:1.27.0-bookworm AS builder ARG MSGVAULT_VERSION=dev diff --git a/docs/setup.md b/docs/setup.md index 6d76c472c..c9bb5aacd 100644 --- a/docs/setup.md +++ b/docs/setup.md @@ -45,7 +45,7 @@ conda install -c conda-forge msgvault ## Build From Source -Requires Go 1.26+, Bun 1.3.14+, and a C/C++ compiler (GCC or Clang). Bun builds +Requires Go 1.27+, Bun 1.3.14+, and a C/C++ compiler (GCC or Clang). Bun builds the browser application embedded in the binary. CGO is required because msgvault uses `mattn/go-sqlite3` (SQLite with FTS5) and `duckdb-go/v2` (Parquet analytics), both of which compile native extensions. diff --git a/flake.nix b/flake.nix index 185305275..7e5c2fdf7 100644 --- a/flake.nix +++ b/flake.nix @@ -25,19 +25,69 @@ let pkgs = nixpkgs.legacyPackages.${system}; - # Pin Go 1.26.6 until nixpkgs-unstable ships it. + # Pin Go 1.27.0 until the locked nixpkgs revision ships it. # Scoped to msgvault only — do NOT export via overlay, that would # invalidate every Go derivation in the transitive closure. - goPinned = pkgs.go_1_26.overrideAttrs (_: rec { - version = "1.26.6"; + goPinned = pkgs.go_1_26.overrideAttrs (old: rec { + version = "1.27.0"; src = pkgs.fetchurl { url = "https://go.dev/dl/go${version}.src.tar.gz"; - hash = "sha256-oHIcVMaIkBRI13rZs+x+p8R0cwdV/4kTgukuy5P/LLE="; + hash = "sha256-cAJAPXzERSnvbSb2mkSBgmM5Xq18FsBaWAiuBH6+sOU="; }; + patches = + builtins.filter ( + patch: !(pkgs.lib.hasSuffix "go_no_vendor_checks-1.26.patch" (toString patch)) + ) old.patches + ++ [ + (pkgs.fetchurl { + url = "https://raw.githubusercontent.com/NixOS/nixpkgs/67a70befab1966b026701e8e94cf19b1543075c5/pkgs/development/compilers/go/go_no_vendor_checks-1.27.patch"; + hash = "sha256-aTpc6kAX9bAyMMAHqzldcb2EEseCpke7QlJuQ1Bk6jc="; + }) + ]; }); buildGoModule = pkgs.buildGoModule.override { go = goPinned; }; + goplsPinned = (pkgs.gopls.override { buildGoLatestModule = buildGoModule; }).overrideAttrs (_: rec { + version = "0.23.0"; + src = pkgs.fetchFromGitHub { + owner = "golang"; + repo = "tools"; + tag = "gopls/v${version}"; + hash = "sha256-GTRZ0tS2a7Cx4qRf6PfxhkGVPYRoLYOmE+W/2x9Pttk="; + }; + vendorHash = "sha256-rvm33C3z3T6moeEQ4C7aG+dT8ROqmpBFehIpwGFZMrU="; + }); + + golangciLintPinned = + (pkgs.golangci-lint.override { buildGo126Module = buildGoModule; }).overrideAttrs + (_: rec { + version = "2.13.1"; + src = pkgs.fetchFromGitHub { + owner = "golangci"; + repo = "golangci-lint"; + tag = "v${version}"; + hash = "sha256-8nWHSMAwIILfKMPfxWKMimxWt9N+kUsZEAaoAOPbRBE="; + }; + vendorHash = "sha256-yZRqfht5rY2yyoZNtYttE57sB7EYjk71yrKw8dLYzNk="; + }); + + delvePinned = (pkgs.delve.override { inherit buildGoModule; }).overrideAttrs (_: rec { + version = "1.27.1"; + src = pkgs.fetchFromGitHub { + owner = "go-delve"; + repo = "delve"; + tag = "v${version}"; + hash = "sha256-H91QnLyqywgoc3zdTaclzzUxVPagNnxLzKub2gnL25w="; + }; + checkFlags = [ "-skip=TestGeneratedDoc|TestTypecheckRPC" ]; + }); + + gotoolsPinned = pkgs.gotools.override { + inherit buildGoModule; + go = goPinned; + }; + bunPinned = pkgs.bun.overrideAttrs (old: rec { version = "1.3.14"; src = passthru.sources.${system}; @@ -91,10 +141,10 @@ devShells.default = pkgs.mkShell { packages = [ goPinned - pkgs.gopls - pkgs.gotools - pkgs.golangci-lint - pkgs.delve + goplsPinned + gotoolsPinned + golangciLintPinned + delvePinned pkgs.gcc pkgs.prek pkgs.sqlite-interactive diff --git a/go.mod b/go.mod index 758c90604..f3933071f 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module go.kenn.io/msgvault -go 1.26.6 +go 1.27.0 replace github.com/emersion/go-imap/v2 => github.com/hstern/go-imap/v2 v2.0.0-beta.8.0.20260621192506-dabdeca47dc7 @@ -30,7 +30,7 @@ require ( github.com/jhillyerd/enmime v1.3.0 github.com/mattn/go-isatty v0.0.22 github.com/mattn/go-runewidth v0.0.24 - github.com/mattn/go-sqlite3 v1.14.47 + github.com/mattn/go-sqlite3 v1.14.50 github.com/modelcontextprotocol/go-sdk v1.7.0 github.com/mooijtech/go-pst/v6 v6.0.2 github.com/robfig/cron/v3 v3.0.1 diff --git a/go.sum b/go.sum index 4cb07df78..7a231f2ca 100644 --- a/go.sum +++ b/go.sum @@ -212,8 +212,8 @@ github.com/mattn/go-isatty v0.0.22/go.mod h1:ZXfXG4SQHsB/w3ZeOYbR0PrPwLy+n6xiMrJ github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI= github.com/mattn/go-runewidth v0.0.24 h1:cpokDiIn0MGnhdHwuWnJBITySJ20QyNGnY2kR/ay2DU= github.com/mattn/go-runewidth v0.0.24/go.mod h1:XBkDxAl56ILZc9knddidhrOlY5R/pDhgLpndooCuJAs= -github.com/mattn/go-sqlite3 v1.14.47 h1:jOBI62gS7nKeZv+as1oGEy0+1qISgXwH/QBlR6KbfIo= -github.com/mattn/go-sqlite3 v1.14.47/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w= +github.com/mattn/go-sqlite3 v1.14.50 h1:dmdFvo1XG4MPzA4IkAmE9upVz/Nj31uRoM5+jC8hYbY= +github.com/mattn/go-sqlite3 v1.14.50/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w= github.com/microcosm-cc/bluemonday v1.0.27 h1:MpEUotklkwCSLeH+Qdx1VJgNqLlpY2KXwXFM08ygZfk= github.com/microcosm-cc/bluemonday v1.0.27/go.mod h1:jFi9vgW+H7c3V0lb6nR74Ib/DIB5OBs92Dimizgw2cA= github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 h1:AMFGa4R4MiIpspGNG7Z948v4n35fFGB3RR3G/ry4FWs= diff --git a/internal/activity/projector.go b/internal/activity/projector.go index 4d9e5272d..0653b00d4 100644 --- a/internal/activity/projector.go +++ b/internal/activity/projector.go @@ -321,8 +321,7 @@ func (p *Projector) RecomputeStale(ctx context.Context) (RunResult, error) { } if err := p.store.RecomputeContactStateContext( ctx, personIDs, current); err != nil { - var stale *store.ErrActivityProjectionStale - if errors.As(err, &stale) { + if _, ok := errors.AsType[*store.ErrActivityProjectionStale](err); ok { return result, errProjectionEpochChanged } return result, err @@ -395,8 +394,7 @@ func (p *Projector) convergeTimezone( } if err := p.store.CompleteActivityTimezoneTransitionContext( ctx, current); err != nil { - var stale *store.ErrActivityProjectionStale - if errors.As(err, &stale) { + if _, ok := errors.AsType[*store.ErrActivityProjectionStale](err); ok { continue } return err @@ -404,8 +402,7 @@ func (p *Projector) convergeTimezone( if direct.Active { if err := p.store.CompleteActivityDirectLimitTransitionContext( ctx, direct); err != nil { - var stale *store.ErrActivityProjectionStale - if errors.As(err, &stale) { + if _, ok := errors.AsType[*store.ErrActivityProjectionStale](err); ok { continue } return err @@ -462,8 +459,7 @@ func (p *Projector) convergeDirectLimit( } if err := p.store.CompleteActivityDirectLimitTransitionContext( ctx, current); err != nil { - var stale *store.ErrActivityProjectionStale - if errors.As(err, &stale) { + if _, ok := errors.AsType[*store.ErrActivityProjectionStale](err); ok { continue } return err @@ -557,8 +553,7 @@ func (p *Projector) reconcileRevisions( } if err := p.store.CompareAndSetActivityReconciledRevisionsContext( ctx, target); err != nil { - var stale *store.ErrActivityProjectionStale - if errors.As(err, &stale) { + if _, ok := errors.AsType[*store.ErrActivityProjectionStale](err); ok { continue } return err @@ -668,8 +663,7 @@ func (p *Projector) projectCandidates( p.advanceWatermark(ctx, current, result) return nil } - var stale *store.ErrActivityProjectionStale - if !errors.As(err, &stale) { + if _, ok := errors.AsType[*store.ErrActivityProjectionStale](err); !ok { return err } if attempt+1 == projectorStaleRetries { diff --git a/internal/api/activity_routes.go b/internal/api/activity_routes.go index 5aa5a813c..0b297993a 100644 --- a/internal/api/activity_routes.go +++ b/internal/api/activity_routes.go @@ -464,8 +464,7 @@ func (s *Server) handleCreateDayEntry(w http.ResponseWriter, r *http.Request) { } body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxDailyNoteRequestBytes)) if err != nil { - var maxBytes *http.MaxBytesError - if errors.As(err, &maxBytes) { + if _, ok := errors.AsType[*http.MaxBytesError](err); ok { writeError(w, http.StatusRequestEntityTooLarge, "request_too_large", "daily entry request is too large") return diff --git a/internal/api/cli_handlers.go b/internal/api/cli_handlers.go index fc38d8bf3..25ef15879 100644 --- a/internal/api/cli_handlers.go +++ b/internal/api/cli_handlers.go @@ -1768,8 +1768,7 @@ func (s *Server) cliDedupDeleteError(err error) *apiHTTPError { if err == nil { return nil } - var requestErr *cliRequestError - if errors.As(err, &requestErr) { + if requestErr, ok := errors.AsType[*cliRequestError](err); ok { return newAPIHTTPError(http.StatusBadRequest, requestErr.code, requestErr.message) } s.logger.Error("failed CLI dedup delete operation", "error", err) diff --git a/internal/api/explore_review_test.go b/internal/api/explore_review_test.go index d32031f5e..2c2919bd4 100644 --- a/internal/api/explore_review_test.go +++ b/internal/api/explore_review_test.go @@ -1699,7 +1699,7 @@ func nextExploreCursor(t *testing.T, response *httptest.ResponseRecorder) string func tamperExploreCursor(t *testing.T, encoded, field string, value any) string { t.Helper() - payload := strings.SplitN(encoded, ".", 2)[0] + payload, _, _ := strings.Cut(encoded, ".") data, err := base64.RawURLEncoding.DecodeString(payload) require.NoError(t, err) var cursor map[string]any diff --git a/internal/api/handlers.go b/internal/api/handlers.go index 3303ee8b4..a413617a1 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -479,29 +479,27 @@ func messageDetailFromQuery(qMsg *query.MessageDetail) MessageDetail { } return MessageDetail{ - MessageSummary: MessageSummary{ - ID: qMsg.ID, - SourceID: qMsg.SourceID, - SourceMessageID: qMsg.SourceMessageID, - ConversationID: qMsg.ConversationID, - Subject: qMsg.Subject, - MessageType: qMsg.MessageType, - From: from, - FromEmail: fromEmail, - FromName: fromName, - To: toAddrs, - Cc: ccAddrs, - Bcc: bccAddrs, - SentAt: qMsg.SentAt.UTC().Format(time.RFC3339), - DeletedAt: formatDeletedAt(qMsg.DeletedAt), - Snippet: qMsg.Snippet, - Labels: labels, - HasAttach: qMsg.HasAttachments, - SizeBytes: qMsg.SizeEstimate, - }, - Body: body, - BodyHTML: qMsg.BodyHTML, - Attachments: attachments, + ID: qMsg.ID, + SourceID: qMsg.SourceID, + SourceMessageID: qMsg.SourceMessageID, + ConversationID: qMsg.ConversationID, + Subject: qMsg.Subject, + MessageType: qMsg.MessageType, + From: from, + FromEmail: fromEmail, + FromName: fromName, + To: toAddrs, + Cc: ccAddrs, + Bcc: bccAddrs, + SentAt: qMsg.SentAt.UTC().Format(time.RFC3339), + DeletedAt: formatDeletedAt(qMsg.DeletedAt), + Snippet: qMsg.Snippet, + Labels: labels, + HasAttach: qMsg.HasAttachments, + SizeBytes: qMsg.SizeEstimate, + Body: body, + BodyHTML: qMsg.BodyHTML, + Attachments: attachments, } } diff --git a/internal/api/organizations.go b/internal/api/organizations.go index 6c78c1af8..35b33eef4 100644 --- a/internal/api/organizations.go +++ b/internal/api/organizations.go @@ -718,8 +718,7 @@ func organizationAttributeTarget(w http.ResponseWriter, r *http.Request) (int64, func decodeOrganizationProfileRequest(w http.ResponseWriter, r *http.Request, target any) bool { body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, MaxOrganizationProfileRequestBytes)) if err != nil { - var maxBytesError *http.MaxBytesError - if errors.As(err, &maxBytesError) { + if _, ok := errors.AsType[*http.MaxBytesError](err); ok { writeError(w, http.StatusRequestEntityTooLarge, "organization_profile_too_large", "Organization profile request is too large") return false diff --git a/internal/api/organizations_test.go b/internal/api/organizations_test.go index 897312aa7..31d755b7b 100644 --- a/internal/api/organizations_test.go +++ b/internal/api/organizations_test.go @@ -480,8 +480,8 @@ func TestOrganizationHTTPProfileAcceptsInlineMediaBeyondGenericRequestLimit(t *t require.NoError(json.Unmarshal(createdResponse.Body.Bytes(), &created)) profileBody, err := json.Marshal(OrganizationProfileBody{Media: []OrganizationMediaBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", Data: bytes.Repeat([]byte("x"), 800*1024), + Source: string(store.ProvenanceUser), + MediaKind: "logo", Data: bytes.Repeat([]byte("x"), 800*1024), }}}) require.NoError(err) assert.Greater(len(profileBody), 1<<20, "regression body must exceed the generic decoder limit") @@ -509,9 +509,7 @@ func TestOrganizationHTTPProfileRejectsTooManyValues(t *testing.T) { categories := make([]OrganizationCategoryBody, store.MaxOrganizationProfileValues+1) for i := range categories { categories[i] = OrganizationCategoryBody{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{ - Source: string(store.ProvenanceUser), - }, + Source: string(store.ProvenanceUser), Category: fmt.Sprintf("category-%d", i), } } @@ -538,15 +536,13 @@ func TestOrganizationHTTPProfilePutRoundTripsEnvelopeMetadata(t *testing.T) { activeFrom := time.Date(2024, 3, 1, 12, 0, 0, 0, time.UTC) name := OrganizationNameBody{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{ - TypeLabel: new("work"), - TypeTokens: []string{"work", "primary"}, - Source: string(store.ProvenanceExtraction), - SourceRef: new("message:synthetic-1"), - Confidence: new(0.75), - ActiveFrom: &activeFrom, - }, - Name: "Example Organisation", NameKind: "alias", + TypeLabel: new("work"), + TypeTokens: []string{"work", "primary"}, + Source: string(store.ProvenanceExtraction), + SourceRef: new("message:synthetic-1"), + Confidence: new(0.75), + ActiveFrom: &activeFrom, + Name: "Example Organisation", NameKind: "alias", } firstBody, err := json.Marshal(OrganizationProfileBody{Names: []OrganizationNameBody{name}}) require.NoError(err) @@ -569,8 +565,8 @@ func TestOrganizationHTTPProfilePutRoundTripsEnvelopeMetadata(t *testing.T) { secondBody, err := json.Marshal(OrganizationProfileBody{ Names: []OrganizationNameBody{name}, Categories: []OrganizationCategoryBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - Category: "vendor", + Source: string(store.ProvenanceUser), + Category: "vendor", }}, }) require.NoError(err) @@ -604,12 +600,12 @@ func TestOrganizationHTTPProfileMediaContentRoundTrip(t *testing.T) { logo := []byte("synthetic-logo-bytes") profileBody, err := json.Marshal(OrganizationProfileBody{Media: []OrganizationMediaBody{ { - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", MediaType: new("image/png"), Data: logo, + Source: string(store.ProvenanceUser), + MediaKind: "logo", MediaType: new("image/png"), Data: logo, }, { - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Ordinal: new(1), Source: string(store.ProvenanceUser)}, - MediaKind: "photo", URI: new("https://example.com/photo.png"), + Ordinal: new(1), Source: string(store.ProvenanceUser), + MediaKind: "photo", URI: new("https://example.com/photo.png"), }, }}) require.NoError(err) @@ -665,8 +661,8 @@ func TestOrganizationHTTPProfilePutRetainsInlineMediaViaContentHash(t *testing.T logo := []byte("synthetic-logo-bytes") uri := "https://example.com/logo.png" firstBody, err := json.Marshal(OrganizationProfileBody{Media: []OrganizationMediaBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", MediaType: new("image/png"), URI: &uri, Data: logo, + Source: string(store.ProvenanceUser), + MediaKind: "logo", MediaType: new("image/png"), URI: &uri, Data: logo, }}}) require.NoError(err) firstPut := organizationRequest(t, srv, http.MethodPut, @@ -684,13 +680,13 @@ func TestOrganizationHTTPProfilePutRetainsInlineMediaViaContentHash(t *testing.T // update re-sending it must keep the stored inline content. secondBody, err := json.Marshal(OrganizationProfileBody{ Media: []OrganizationMediaBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", MediaType: new("image/png"), URI: &uri, + Source: string(store.ProvenanceUser), + MediaKind: "logo", MediaType: new("image/png"), URI: &uri, ContentHash: firstProfile.Media[0].ContentHash, }}, Categories: []OrganizationCategoryBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - Category: "vendor", + Source: string(store.ProvenanceUser), + Category: "vendor", }}, }) require.NoError(err) @@ -715,8 +711,8 @@ func TestOrganizationHTTPProfilePutRetainsInlineMediaViaContentHash(t *testing.T // A retention hash that matches no active row is a client error, not a // silent hash-without-bytes insert. staleBody, err := json.Marshal(OrganizationProfileBody{Media: []OrganizationMediaBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", MediaType: new("image/png"), URI: &uri, + Source: string(store.ProvenanceUser), + MediaKind: "logo", MediaType: new("image/png"), URI: &uri, ContentHash: new("0000000000000000000000000000000000000000000000000000000000000000"), }}}) require.NoError(err) @@ -739,8 +735,8 @@ func TestOrganizationHTTPProfilePutEditsInlineMediaMetadataViaContentHash(t *tes logo := []byte("metadata-edit-logo-bytes") firstBody, err := json.Marshal(OrganizationProfileBody{Media: []OrganizationMediaBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", MediaType: new("image/png"), Data: logo, + Source: string(store.ProvenanceUser), + MediaKind: "logo", MediaType: new("image/png"), Data: logo, }}}) require.NoError(err) firstPut := organizationRequest(t, srv, http.MethodPut, @@ -756,8 +752,8 @@ func TestOrganizationHTTPProfilePutEditsInlineMediaMetadataViaContentHash(t *tes // stored bytes into the replacement row, not fail for lack of data. uri := "https://example.com/logo.webp" editBody, err := json.Marshal(OrganizationProfileBody{Media: []OrganizationMediaBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", MediaType: new("image/webp"), URI: &uri, + Source: string(store.ProvenanceUser), + MediaKind: "logo", MediaType: new("image/webp"), URI: &uri, ContentHash: firstProfile.Media[0].ContentHash, }}}) require.NoError(err) @@ -795,8 +791,8 @@ func TestOrganizationHTTPProfilePutRetainsInlineOnlyMediaWithoutURI(t *testing.T logo := []byte("inline-only-logo-bytes") firstBody, err := json.Marshal(OrganizationProfileBody{Media: []OrganizationMediaBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", MediaType: new("image/png"), Data: logo, + Source: string(store.ProvenanceUser), + MediaKind: "logo", MediaType: new("image/png"), Data: logo, }}}) require.NoError(err) firstPut := organizationRequest(t, srv, http.MethodPut, @@ -812,13 +808,13 @@ func TestOrganizationHTTPProfilePutRetainsInlineOnlyMediaWithoutURI(t *testing.T secondBody, err := json.Marshal(OrganizationProfileBody{ Media: []OrganizationMediaBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - MediaKind: "logo", MediaType: new("image/png"), + Source: string(store.ProvenanceUser), + MediaKind: "logo", MediaType: new("image/png"), ContentHash: firstProfile.Media[0].ContentHash, }}, Categories: []OrganizationCategoryBody{{ - OrganizationEnvelopeBody: OrganizationEnvelopeBody{Source: string(store.ProvenanceUser)}, - Category: "vendor", + Source: string(store.ProvenanceUser), + Category: "vendor", }}, }) require.NoError(err) diff --git a/internal/api/params.go b/internal/api/params.go index af5d61ad7..45103a37e 100644 --- a/internal/api/params.go +++ b/internal/api/params.go @@ -26,8 +26,7 @@ func newParamError(param, message string) *paramError { // rejectBadParam writes a 400 for an invalid query parameter, using the // parameter-specific error code (invalid_) when err is a paramError. func (s *Server) rejectBadParam(w http.ResponseWriter, err error) { - var pe *paramError - if errors.As(err, &pe) { + if pe, ok := errors.AsType[*paramError](err); ok { writeError(w, http.StatusBadRequest, "invalid_"+pe.param, pe.message) return } @@ -38,8 +37,7 @@ func (s *Server) rejectBadParam(w http.ResponseWriter, err error) { // that surface errors through writeAPIHTTPError. It preserves the // parameter-specific error code so the response envelope matches rejectBadParam. func apiHTTPErrorFromParam(err error) *apiHTTPError { - var pe *paramError - if errors.As(err, &pe) { + if pe, ok := errors.AsType[*paramError](err); ok { return newAPIHTTPError(http.StatusBadRequest, "invalid_"+pe.param, pe.message) } return newAPIHTTPError(http.StatusBadRequest, "invalid_parameter", err.Error()) diff --git a/internal/api/person_profile_values.go b/internal/api/person_profile_values.go index 2e69a2b7d..63fcd16cf 100644 --- a/internal/api/person_profile_values.go +++ b/internal/api/person_profile_values.go @@ -405,8 +405,7 @@ func decodeProfilePatchRequest( var patch store.PersonProfilePatch body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, MaxPersonProfilePatchBytes)) if err != nil { - var maxBytesError *http.MaxBytesError - if errors.As(err, &maxBytesError) { + if _, ok := errors.AsType[*http.MaxBytesError](err); ok { writeError(w, http.StatusRequestEntityTooLarge, "profile_patch_too_large", "Person profile patch is too large") return patch, false diff --git a/internal/api/task_links.go b/internal/api/task_links.go index 344e9fd3b..bd48971a2 100644 --- a/internal/api/task_links.go +++ b/internal/api/task_links.go @@ -127,8 +127,7 @@ func (s *Server) handleCreateOrLinkMessageTask(w http.ResponseWriter, r *http.Re decoder := json.NewDecoder(r.Body) decoder.DisallowUnknownFields() if err := decoder.Decode(&request); err != nil { - var maxBytesErr *http.MaxBytesError - if errors.As(err, &maxBytesErr) { + if _, ok := errors.AsType[*http.MaxBytesError](err); ok { writeError(w, http.StatusRequestEntityTooLarge, "request_too_large", "Task request is too large") return } @@ -169,7 +168,7 @@ func (s *Server) handleListMessageTasks(w http.ResponseWriter, r *http.Request) return } if s.taskLinkOperations == nil { - writeJSON(w, http.StatusOK, taskLinkLookupResponse(identity, tasklinks.LookupResult{IndexStatus: tasklinks.IndexStatus{State: tasklinks.StateUnavailable, Complete: false, Reason: tasklinks.ReasonUnavailable}, Tasks: []tasklinks.TaskSummary{}})) + writeJSON(w, http.StatusOK, taskLinkLookupResponse(identity, tasklinks.LookupResult{State: tasklinks.StateUnavailable, Complete: false, Reason: tasklinks.ReasonUnavailable, Tasks: []tasklinks.TaskSummary{}})) return } writeJSON(w, http.StatusOK, taskLinkLookupResponse(identity, s.taskLinkOperations.Lookup(r.Context(), identity))) diff --git a/internal/calsync/incremental.go b/internal/calsync/incremental.go index 7dce0833e..93108eccc 100644 --- a/internal/calsync/incremental.go +++ b/internal/calsync/incremental.go @@ -129,8 +129,7 @@ func (s *Syncer) incrementalCalendar(ctx context.Context, src *store.Source, cal PageToken: pageToken, }) if err != nil { - var gone *gcal.GoneError - if errors.As(err, &gone) { + if _, ok := errors.AsType[*gcal.GoneError](err); ok { _ = s.store.FailSync(syncID, ErrSyncTokenExpired.Error()) return ErrSyncTokenExpired } diff --git a/internal/daemonclient/client_test.go b/internal/daemonclient/client_test.go index dcba542ca..3ccae360a 100644 --- a/internal/daemonclient/client_test.go +++ b/internal/daemonclient/client_test.go @@ -155,10 +155,10 @@ func TestCLIIdentityDiscoverProviderStreamsConvertedEventsAndRequest(t *testing. requirements.NoError(err) var events []identityops.DiscoverEvent err = client.DiscoverCLIIdentities(t.Context(), identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Apply: true, - Provider: true, - Confirm: []string{"weak@example.test"}, + SourceID: 14, + Apply: true, + Provider: true, + Confirm: []string{"weak@example.test"}, }, func(event identityops.DiscoverEvent) error { events = append(events, event) return nil @@ -195,7 +195,7 @@ func TestCLIIdentityDiscoverRejectsStreamWithoutResult(t *testing.T) { require.NoError(t, err) err = client.DiscoverCLIIdentities(t.Context(), identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, + SourceID: 14, }, nil) require.Error(t, err) @@ -220,7 +220,7 @@ func TestCLIIdentityDiscoverConsumesSanitizedTerminalError(t *testing.T) { var events []identityops.DiscoverEvent err = client.DiscoverCLIIdentities(t.Context(), identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, + SourceID: 14, }, func(event identityops.DiscoverEvent) error { events = append(events, event) return nil @@ -290,7 +290,7 @@ func TestCLIIdentityImportSendsParsedEntriesAndConvertsResult(t *testing.T) { client, err := New(Config{URL: srv.URL, AllowInsecure: true}) requirements.NoError(err) result, err := client.ImportCLIIdentities(t.Context(), identityops.ImportRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, + SourceID: 14, Entries: []identityops.ImportEntry{{ Identifier: "alias@example.test", State: "disabled", }}, diff --git a/internal/discord/catalog.go b/internal/discord/catalog.go index 70ecd4389..18319f6c4 100644 --- a/internal/discord/catalog.go +++ b/internal/discord/catalog.go @@ -460,8 +460,7 @@ func newCatalogIssue(scope CatalogScope, guildID, parentID string, err error) Ca issue.Kind = CatalogIssueMalformedPage return issue } - var apiErr *APIError - if errors.As(err, &apiErr) { + if apiErr, ok := errors.AsType[*APIError](err); ok { issue.StatusCode = apiErr.StatusCode issue.DiscordCode = apiErr.Code switch apiErr.StatusCode { diff --git a/internal/discord/importer.go b/internal/discord/importer.go index 4339a099e..c5bf62552 100644 --- a/internal/discord/importer.go +++ b/internal/discord/importer.go @@ -464,10 +464,10 @@ func (imp *Importer) importerContainers( continue } containers = append(containers, importerContainer{ - CatalogContainer: CatalogContainer{Channel: Channel{ + Channel: Channel{ ID: containerID, GuildID: guildID, MemberCount: storedContainerMemberCount(storedMetadata[containerID]), - }}, + }, preserveMetadata: true, }) } diff --git a/internal/documentindex/reconcile_integration_test.go b/internal/documentindex/reconcile_integration_test.go index 8189aa77e..512b1a080 100644 --- a/internal/documentindex/reconcile_integration_test.go +++ b/internal/documentindex/reconcile_integration_test.go @@ -1,9 +1,12 @@ package documentindex import ( + "context" + "strconv" "strings" "sync" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -186,6 +189,157 @@ func TestConcurrentOccurrenceReconciliationKeepsHighestSourceSequence(t *testing assert.Equal(t, int64(100), sequence) } +func TestSQLiteOccurrenceReconciliationReadsAfterWriterSlot(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + f := storetest.New(t) + if f.Store.IsPostgreSQL() { + t.Skip("SQLite writer-slot regression") + } + messageID := f.CreateMessage("document-reconcile-writer-slot") + attachmentID := createReconcileAttachment(t, f, messageID, "1") + _, eligible, err := f.Store.ReconcileDocumentOccurrence(t.Context(), attachmentID, 9) + require.NoError(err) + require.True(eligible) + + holder, err := f.Store.DB().Conn(t.Context()) + require.NoError(err) + held := true + t.Cleanup(func() { + if held { + _, _ = holder.ExecContext(context.Background(), "ROLLBACK") + } + _ = holder.Close() + }) + _, err = holder.ExecContext(t.Context(), "BEGIN IMMEDIATE") + require.NoError(err) + + type reconcileResult struct { + eligible bool + err error + } + result := make(chan reconcileResult, 1) + go func() { + _, reconciledEligible, reconcileErr := f.Store.ReconcileDocumentOccurrence( + t.Context(), attachmentID, 10, + ) + result <- reconcileResult{eligible: reconciledEligible, err: reconcileErr} + }() + require.Eventually(func() bool { + return f.Store.DB().Stats().InUse >= 2 || len(result) > 0 + }, time.Second, time.Millisecond) + select { + case <-result: + require.Fail("reconciliation returned while the SQLite writer slot was held") + default: + } + require.GreaterOrEqual(f.Store.DB().Stats().InUse, 2) + + select { + case <-result: + require.Fail("reconciliation returned while the SQLite writer slot was held") + case <-time.After(50 * time.Millisecond): + } + _, err = holder.ExecContext(t.Context(), ` + UPDATE attachments SET attachment_role = ? WHERE id = ?`, + store.AttachmentRoleInline, attachmentID) + require.NoError(err) + _, err = holder.ExecContext(t.Context(), ` + DELETE FROM document_occurrences + WHERE attachment_id = ? AND source_sequence <= ?`, attachmentID, 11) + require.NoError(err) + _, err = holder.ExecContext(t.Context(), "COMMIT") + require.NoError(err) + held = false + reconciled := <-result + require.NoError(reconciled.err) + assert.False(reconciled.eligible) + assert.Empty(documentOccurrenceAttachmentIDs(t, f)) +} + +func TestPostgreSQLOccurrenceReconciliationSerializesEligibilityRead(t *testing.T) { + require := require.New(t) + assert := assert.New(t) + f := storetest.New(t) + if !f.Store.IsPostgreSQL() { + t.Skip("PostgreSQL advisory-lock regression") + } + messageID := f.CreateMessage("document-reconcile-advisory-lock") + attachmentID := createReconcileAttachment(t, f, messageID, "2") + _, eligible, err := f.Store.ReconcileDocumentOccurrence(t.Context(), attachmentID, 9) + require.NoError(err) + require.True(eligible) + + holder, err := f.Store.DB().Conn(t.Context()) + require.NoError(err) + lockName := "msgvault.document_occurrence.attachment:" + strconv.FormatInt(attachmentID, 10) + held := true + t.Cleanup(func() { + if held { + _, _ = holder.ExecContext(context.Background(), f.Store.Rebind(` + SELECT pg_advisory_unlock(hashtextextended(CAST(? AS TEXT), 0))`), lockName) + } + _ = holder.Close() + }) + _, err = holder.ExecContext(t.Context(), f.Store.Rebind(` + SELECT pg_advisory_lock(hashtextextended(CAST(? AS TEXT), 0))`), lockName) + require.NoError(err) + var waitingBefore int + require.NoError(f.Store.DB().QueryRow(` + SELECT COUNT(*) FROM pg_locks + WHERE locktype = 'advisory' AND NOT granted`).Scan(&waitingBefore)) + + type reconcileResult struct { + eligible bool + err error + } + lower := make(chan reconcileResult, 1) + go func() { + _, reconciledEligible, reconcileErr := f.Store.ReconcileDocumentOccurrence( + t.Context(), attachmentID, 10, + ) + lower <- reconcileResult{eligible: reconciledEligible, err: reconcileErr} + }() + require.Eventually(func() bool { + var waiting int + err := f.Store.DB().QueryRow(` + SELECT COUNT(*) FROM pg_locks + WHERE locktype = 'advisory' AND NOT granted`).Scan(&waiting) + return err == nil && waiting >= waitingBefore+1 + }, time.Second, time.Millisecond) + + _, err = f.Store.DB().Exec(f.Store.Rebind(` + UPDATE attachments SET attachment_role = ? WHERE id = ?`), + store.AttachmentRoleInline, attachmentID) + require.NoError(err) + higher := make(chan reconcileResult, 1) + go func() { + _, reconciledEligible, reconcileErr := f.Store.ReconcileDocumentOccurrence( + t.Context(), attachmentID, 11, + ) + higher <- reconcileResult{eligible: reconciledEligible, err: reconcileErr} + }() + require.Eventually(func() bool { + var waiting int + err := f.Store.DB().QueryRow(` + SELECT COUNT(*) FROM pg_locks + WHERE locktype = 'advisory' AND NOT granted`).Scan(&waiting) + return len(higher) > 0 || err == nil && waiting >= waitingBefore+2 + }, time.Second, time.Millisecond) + + _, err = holder.ExecContext(t.Context(), f.Store.Rebind(` + SELECT pg_advisory_unlock(hashtextextended(CAST(? AS TEXT), 0))`), lockName) + require.NoError(err) + held = false + lowResult := <-lower + highResult := <-higher + require.NoError(lowResult.err) + require.NoError(highResult.err) + assert.False(lowResult.eligible) + assert.False(highResult.eligible) + assert.Empty(documentOccurrenceAttachmentIDs(t, f)) +} + func TestReconcilerRemovesCascadedOccurrenceFromJournalReplay(t *testing.T) { require := require.New(t) f := storetest.New(t) diff --git a/internal/gmail/client.go b/internal/gmail/client.go index 5bff2e871..02ccba8e0 100644 --- a/internal/gmail/client.go +++ b/internal/gmail/client.go @@ -474,8 +474,7 @@ func (c *Client) GetMessagesRawBatchWithErrors(ctx context.Context, messageIDs [ // Log but don't fail the batch - allow partial results. // 404s are expected (message deleted between history scan and fetch), // so log at debug level to avoid noise during incremental sync. - var nfe *NotFoundError - if errors.As(err, &nfe) { + if _, ok := errors.AsType[*NotFoundError](err); ok { c.logger.Debug("message deleted before fetch", "id", id) } else { c.logger.Warn("failed to fetch message", "id", id, "error", err) diff --git a/internal/granola/importer_test.go b/internal/granola/importer_test.go index b53deb868..64a2b5018 100644 --- a/internal/granola/importer_test.go +++ b/internal/granola/importer_test.go @@ -244,11 +244,9 @@ func TestImport_TranscriptTimestampFallbacksRemainSearchable(t *testing.T) { notes := []*Note{ { - NoteSummary: NoteSummary{ - ID: "sparse-transcript-times", Title: "Sparse transcript times", - Owner: User{Name: "Test User", Email: "user@example.com"}, - CreatedAt: createdAt, UpdatedAt: createdAt, - }, + ID: "sparse-transcript-times", Title: "Sparse transcript times", + Owner: User{Name: "Test User", Email: "user@example.com"}, + CreatedAt: createdAt, UpdatedAt: createdAt, Transcript: []TranscriptSegment{ {Speaker: Speaker{Name: "Untimed"}, Text: "No timestamp"}, {Speaker: Speaker{Name: "Timed"}, Text: "First timestamp", StartTime: firstTranscriptAt}, @@ -256,11 +254,9 @@ func TestImport_TranscriptTimestampFallbacksRemainSearchable(t *testing.T) { }, }, { - NoteSummary: NoteSummary{ - ID: "missing-transcript-times", Title: "Missing transcript times", - Owner: User{Name: "Test User", Email: "user@example.com"}, - CreatedAt: createdAt.Add(time.Hour), UpdatedAt: createdAt.Add(time.Hour), - }, + ID: "missing-transcript-times", Title: "Missing transcript times", + Owner: User{Name: "Test User", Email: "user@example.com"}, + CreatedAt: createdAt.Add(time.Hour), UpdatedAt: createdAt.Add(time.Hour), Transcript: []TranscriptSegment{{ Speaker: Speaker{Name: "Untimed"}, Text: "Created-at fallback", }}, diff --git a/internal/identityops/discovery_test.go b/internal/identityops/discovery_test.go index 5fe1890ea..7e6c495a4 100644 --- a/internal/identityops/discovery_test.go +++ b/internal/identityops/discovery_test.go @@ -233,7 +233,7 @@ func TestDiscoverWithProviderExternalEvidencePreviewAndApplyUseSameMergedCandida }} preview, err := identityops.DiscoverWithExternalEvidence(t.Context(), previewStore, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, + SourceID: 14, }, evidence, nil) requirements.NoError(err) @@ -241,7 +241,7 @@ func TestDiscoverWithProviderExternalEvidencePreviewAndApplyUseSameMergedCandida applyStore.pages = previewStore.pages applyStore.identities = previewStore.identities apply, err := identityops.DiscoverWithExternalEvidence(t.Context(), applyStore, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, Apply: true, + SourceID: 14, Apply: true, }, evidence, nil) requirements.NoError(err) @@ -355,7 +355,7 @@ func TestDiscoverClassifiesStrongWeakConfirmedAndCaseVariants(t *testing.T) { }} got, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, + SourceID: 14, }, nil) require.NoError(err) require.Len(got.Candidates, 3) @@ -380,8 +380,8 @@ func TestDiscoverApplyNeverConfirmsRecipientOnlyCandidate(t *testing.T) { }} _, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Apply: true, + SourceID: 14, + Apply: true, }, nil) require.NoError(err) assert.Equal([]int64{14}, st.batchSourceIDs) @@ -405,8 +405,8 @@ func TestDiscoverApplyDoesNotConfirmAmbiguousFromAddresses(t *testing.T) { }} got, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Apply: true, + SourceID: 14, + Apply: true, }, nil) require.NoError(err) require.Len(got.Candidates, 2) @@ -435,7 +435,7 @@ func TestDiscoverApplyMergesOnlyNewStrongSignalsIntoConfirmedCandidate(t *testin }} _, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, Apply: true, + SourceID: 14, Apply: true, }, nil) require.NoError(err) assert.Equal([]int64{14}, st.batchSourceIDs) @@ -455,7 +455,7 @@ func TestDiscoverApplyMergesOnlyNewStrongSignalsIntoConfirmedCandidate(t *testin }}, }} _, err = identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, Apply: true, + SourceID: 14, Apply: true, }, nil) require.NoError(err) assert.Empty(st.batchCalls, "unchanged evidence must not start an empty write batch") @@ -475,9 +475,9 @@ func TestDiscoverExplicitlyConfirmsOnlyCompletedWeakCandidate(t *testing.T) { }} got, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Apply: true, - Confirm: []string{"WEAK@example.test", "weak@example.test"}, + SourceID: 14, + Apply: true, + Confirm: []string{"WEAK@example.test", "weak@example.test"}, }, nil) require.NoError(err) require.Len(got.Applied, 2) @@ -495,9 +495,9 @@ func TestDiscoverExplicitlyConfirmsOnlyCompletedWeakCandidate(t *testing.T) { }, }} _, err = identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Apply: true, - Confirm: []string{"strong@example.test"}, + SourceID: 14, + Apply: true, + Confirm: []string{"strong@example.test"}, }, nil) require.ErrorContains(err, "weak candidate") assert.Empty(st.batchCalls, "all explicit confirmations must validate before writing") @@ -513,8 +513,8 @@ func TestDiscoverConfirmRequiresApply(t *testing.T) { }} _, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Confirm: []string{"weak@example.test"}, + SourceID: 14, + Confirm: []string{"weak@example.test"}, }, nil) require.ErrorContains(t, err, "requires apply") assert.Empty(t, st.batchCalls) @@ -696,7 +696,7 @@ func TestDiscoverRejectsUnsafeAddressesAndCountsDistinctMessages(t *testing.T) { }} got, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, + SourceID: 14, }, nil) require.NoError(err) require.Len(got.Candidates, 1) @@ -734,7 +734,7 @@ func TestDiscoverCancellationAfterPageWritesNothing(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) got, err := identityops.Discover(ctx, st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, Apply: true, PageSize: 1, + SourceID: 14, Apply: true, PageSize: 1, }, func(identityops.DiscoverProgress) error { cancel() return nil @@ -763,7 +763,7 @@ func TestDiscoverInterruptedApplyRerunMatchesOneShot(t *testing.T) { return outcomes, context.Canceled } _, err := identityops.Discover(t.Context(), resumed, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, Apply: true, + SourceID: 14, Apply: true, }, nil) require.ErrorIs(err, context.Canceled) @@ -772,7 +772,7 @@ func TestDiscoverInterruptedApplyRerunMatchesOneShot(t *testing.T) { return mergeConfirmationState(resumedState, confirmations), nil } _, err = identityops.Discover(t.Context(), resumed, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, Apply: true, + SourceID: 14, Apply: true, }, nil) require.NoError(err) @@ -783,7 +783,7 @@ func TestDiscoverInterruptedApplyRerunMatchesOneShot(t *testing.T) { return mergeConfirmationState(cleanState, confirmations), nil } _, err = identityops.Discover(t.Context(), clean, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, Apply: true, + SourceID: 14, Apply: true, }, nil) require.NoError(err) assert.Equal(cleanState, resumedState) @@ -810,7 +810,7 @@ func TestDiscoverApplyErrorReturnsCommittedPrefix(t *testing.T) { } got, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, Apply: true, + SourceID: 14, Apply: true, }, nil) requirements.ErrorIs(err, applyErr) @@ -888,8 +888,8 @@ func TestDiscoverApplyAfterParticipantMergeConfirmsOnlyEnvelopeAddress(t *testin require.NoError(st.MergeParticipants(aliceID, bobID), "merge alice into bob") result, err := identityops.Discover(t.Context(), st, identityops.DiscoverRequest{ - SourceSelector: identityops.SourceSelector{SourceID: source.ID}, - Apply: true, + SourceID: source.ID, + Apply: true, }, nil) require.NoError(err, "discover with apply") diff --git a/internal/identityops/import_test.go b/internal/identityops/import_test.go index 4179e9e94..2153a6958 100644 --- a/internal/identityops/import_test.go +++ b/internal/identityops/import_test.go @@ -80,8 +80,8 @@ func TestIdentityImportPreviewAndApplyShareDeterministicCandidates(t *testing.T) }} preview, err := identityops.Import(t.Context(), previewStore, identityops.ImportRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Entries: entries, + SourceID: 14, + Entries: entries, }) requirements.NoError(err) assertions.Empty(previewStore.batchCalls) @@ -97,9 +97,9 @@ func TestIdentityImportPreviewAndApplyShareDeterministicCandidates(t *testing.T) applyStore := newDiscoveryFakeStore() applyStore.identities = previewStore.identities apply, err := identityops.Import(t.Context(), applyStore, identityops.ImportRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Entries: entries, - Apply: true, + SourceID: 14, + Entries: entries, + Apply: true, }) requirements.NoError(err) assertions.Equal(preview.Candidates, apply.Candidates) @@ -141,10 +141,10 @@ func TestIdentityImportValidatesEveryRowAndSignalBeforeWriting(t *testing.T) { t.Run(test.name, func(t *testing.T) { st := newDiscoveryFakeStore() _, err := identityops.Import(t.Context(), st, identityops.ImportRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Entries: test.entries, - Signal: test.signal, - Apply: true, + SourceID: 14, + Entries: test.entries, + Signal: test.signal, + Apply: true, }) require.ErrorContains(t, err, test.want) @@ -164,7 +164,7 @@ func TestIdentityImportApplyIsSourceScopedStateIndependentAndIdempotent(t *testi requirements.NoError(st.AddAccountIdentity(source.ID, "Old@Example.test", "manual")) requirements.NoError(st.AddAccountIdentity(other.ID, "other-alias@example.test", "manual")) req := identityops.ImportRequest{ - SourceSelector: identityops.SourceSelector{SourceID: source.ID}, + SourceID: source.ID, Entries: []identityops.ImportEntry{ {Identifier: "old@example.test", State: "deleted"}, {Identifier: "waiting@example.test", State: "pending"}, @@ -197,10 +197,10 @@ func TestIdentityImportMergesAdditionalSignalIdempotently(t *testing.T) { requirements.NoError(err) requirements.NoError(st.AddAccountIdentity(source.ID, "Alias@Example.test", "manual")) req := identityops.ImportRequest{ - SourceSelector: identityops.SourceSelector{SourceID: source.ID}, - Entries: []identityops.ImportEntry{{Identifier: "alias@example.test", State: "deleted"}}, - Signal: "bulk-import", - Apply: true, + SourceID: source.ID, + Entries: []identityops.ImportEntry{{Identifier: "alias@example.test", State: "deleted"}}, + Signal: "bulk-import", + Apply: true, } first, err := identityops.Import(t.Context(), st, req) @@ -225,9 +225,9 @@ func TestIdentityImportCancellationPreventsWrites(t *testing.T) { st := newDiscoveryFakeStore() _, err := identityops.Import(ctx, st, identityops.ImportRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, - Entries: []identityops.ImportEntry{{Identifier: "alias@example.test"}}, - Apply: true, + SourceID: 14, + Entries: []identityops.ImportEntry{{Identifier: "alias@example.test"}}, + Apply: true, }) require.ErrorIs(t, err, context.Canceled) @@ -246,7 +246,7 @@ func TestIdentityImportReturnsCommittedPrefixOnBatchError(t *testing.T) { } result, err := identityops.Import(t.Context(), st, identityops.ImportRequest{ - SourceSelector: identityops.SourceSelector{SourceID: 14}, + SourceID: 14, Entries: []identityops.ImportEntry{ {Identifier: "first@example.test"}, {Identifier: "second@example.test"}, diff --git a/internal/jsonexact/jsonexact.go b/internal/jsonexact/jsonexact.go index db9b1aeb6..d29e3498f 100644 --- a/internal/jsonexact/jsonexact.go +++ b/internal/jsonexact/jsonexact.go @@ -77,7 +77,7 @@ func jsonFields( if !field.IsExported() { continue } - name := strings.Split(field.Tag.Get("json"), ",")[0] + name, _, _ := strings.Cut(field.Tag.Get("json"), ",") switch name { case "-": continue diff --git a/internal/mcp/contract.go b/internal/mcp/contract.go index c1474552d..acfa8563d 100644 --- a/internal/mcp/contract.go +++ b/internal/mcp/contract.go @@ -41,8 +41,7 @@ func (e *internalError) Unwrap() error { } func newInternalError(operation string, err error) error { - var privateErr *internalError - if errors.As(err, &privateErr) { + if _, ok := errors.AsType[*internalError](err); ok { return err } return &internalError{operation: operation, cause: err} diff --git a/internal/mcp/handlers.go b/internal/mcp/handlers.go index 77cf8aa13..0637c3575 100644 --- a/internal/mcp/handlers.go +++ b/internal/mcp/handlers.go @@ -464,8 +464,7 @@ func translateDaemonRequestError(err error) *toolResult { } func dependencyError(operation string, err error) (*toolResult, error) { - var expected *expectedHandlerError - if errors.As(err, &expected) { + if expected, ok := errors.AsType[*expectedHandlerError](err); ok { return toolErrorResult(expected.message), nil } if result := translateVectorErr(err); result != nil { @@ -1787,8 +1786,7 @@ func (h *handlers) getAttachment(ctx context.Context, req toolRequest) (*toolRes payload, err := h.attachmentService().load(ctx, id) if err != nil { - var unavailable *attachmentUnavailableError - if errors.As(err, &unavailable) { + if unavailable, ok := errors.AsType[*attachmentUnavailableError](err); ok { return toolErrorResult(unavailable.message), nil } return nil, err @@ -1822,8 +1820,7 @@ func (h *handlers) exportAttachment(ctx context.Context, req toolRequest) (*tool payload, err := h.attachmentService().load(ctx, id) if err != nil { - var unavailable *attachmentUnavailableError - if errors.As(err, &unavailable) { + if unavailable, ok := errors.AsType[*attachmentUnavailableError](err); ok { return toolErrorResult(unavailable.message), nil } return nil, err @@ -2001,7 +1998,7 @@ func (h *handlers) aggregate(ctx context.Context, req toolRequest) (*toolResult, } viewTypeMap := map[string]query.ViewType{ - "sender": query.ViewSenders, //nolint:goconst // Stable public enum value shared with the MCP schema. + "sender": query.ViewSenders, "recipient": query.ViewRecipients, "domain": query.ViewDomains, "label": query.ViewLabels, diff --git a/internal/mcp/middleware.go b/internal/mcp/middleware.go index 9a7117d86..6650aa39e 100644 --- a/internal/mcp/middleware.go +++ b/internal/mcp/middleware.go @@ -60,8 +60,7 @@ func errorIsolationMiddleware(next sdkmcp.MethodHandler) sdkmcp.MethodHandler { if err == nil { return result, nil } - var protocolErr *jsonrpc.Error - if errors.As(err, &protocolErr) { + if _, ok := errors.AsType[*jsonrpc.Error](err); ok { return result, err } slog.Error("MCP method failed with unexpected error", "method", method, "error", err) diff --git a/internal/mcp/resources.go b/internal/mcp/resources.go index 7c0389626..e4d65b04c 100644 --- a/internal/mcp/resources.go +++ b/internal/mcp/resources.go @@ -166,8 +166,7 @@ func registerAttachmentResources(server *sdkmcp.Server, h *handlers) { } payload, err := h.attachmentService().load(ctx, id) if err != nil { - var unavailable *attachmentUnavailableError - if errors.As(err, &unavailable) { + if _, ok := errors.AsType[*attachmentUnavailableError](err); ok { return nil, sdkmcp.ResourceNotFoundError(rawURI) } return nil, mapInternalError(err) diff --git a/internal/mcp/server.go b/internal/mcp/server.go index 78efea2d2..ae8a554af 100644 --- a/internal/mcp/server.go +++ b/internal/mcp/server.go @@ -136,8 +136,7 @@ func officialToolHandler( } func mapInternalError(err error) error { - var privateErr *internalError - if errors.As(err, &privateErr) { + if privateErr, ok := errors.AsType[*internalError](err); ok { slog.Error("MCP operation failed", "operation", privateErr.operation, "error", privateErr.cause) } else { slog.Error("MCP operation failed with unclassified error", "error", err) diff --git a/internal/mcp/server_test.go b/internal/mcp/server_test.go index 56d3826a1..347c386b7 100644 --- a/internal/mcp/server_test.go +++ b/internal/mcp/server_test.go @@ -1431,7 +1431,7 @@ func TestAttachVectorChunkMatches_HTMLOnlyUsesEmbeddingCorpus(t *testing.T) { backend: backend, vectorCfg: vector.Config{}, } - items := []searchMessageItem{{MessageSummary: query.MessageSummary{ID: messageID}}} + items := []searchMessageItem{{ID: messageID}} require.NoError(t, h.attachVectorChunkMatches(context.Background(), 1, []float32{1}, items, 0)) diff --git a/internal/oauth/readonly_scopes_test.go b/internal/oauth/readonly_scopes_test.go index 388201e18..d9d593d55 100644 --- a/internal/oauth/readonly_scopes_test.go +++ b/internal/oauth/readonly_scopes_test.go @@ -196,9 +196,10 @@ func TestTokenIssuedByDifferentClient(t *testing.T) { if tt.writeFile { tf := tokenFile{ - Token: oauth2.Token{AccessToken: "t", TokenType: "Bearer"}, - Scopes: Scopes, - ClientID: tt.clientID, + AccessToken: "t", + TokenType: "Bearer", + Scopes: Scopes, + ClientID: tt.clientID, } data, err := json.Marshal(tf) require.NoError(t, err) diff --git a/internal/opserr/opserr.go b/internal/opserr/opserr.go index 425f9298d..fc2a85cb7 100644 --- a/internal/opserr/opserr.go +++ b/internal/opserr/opserr.go @@ -33,8 +33,7 @@ func (e *Error) Unwrap() error { // KindOf returns the operation error kind, defaulting to internal errors. func KindOf(err error) Kind { - var opErr *Error - if errors.As(err, &opErr) { + if opErr, ok := errors.AsType[*Error](err); ok { return opErr.Kind } return KindInternal diff --git a/internal/store/activity_test.go b/internal/store/activity_test.go index a2785d929..ff8c2beb4 100644 --- a/internal/store/activity_test.go +++ b/internal/store/activity_test.go @@ -2256,8 +2256,7 @@ func TestProjectActivityBatchReservesAbsentLegacyQueueAtomically(t *testing.T) { successes++ continue } - var stale *store.ErrActivityProjectionStale - if errors.As(err, &stale) { + if _, ok := errors.AsType[*store.ErrActivityProjectionStale](err); ok { staleErrors++ continue } @@ -2303,8 +2302,7 @@ func TestProjectActivityBatchConcurrentSameTokenCommitsOnce(t *testing.T) { successes++ continue } - var stale *store.ErrActivityProjectionStale - if errors.As(err, &stale) { + if _, ok := errors.AsType[*store.ErrActivityProjectionStale](err); ok { staleErrors++ continue } diff --git a/internal/store/attachment_changes.go b/internal/store/attachment_changes.go index bf4e4ec03..9a806e1a7 100644 --- a/internal/store/attachment_changes.go +++ b/internal/store/attachment_changes.go @@ -226,6 +226,16 @@ func (s *Store) AdvanceAttachmentChangeConsumer( } return s.withTxContext(ctx, func(tx *loggedTx) error { q := boundQuerier{ctx: ctx, q: tx} + if !s.IsPostgreSQL() { + // Reserve SQLite's writer slot before reading the cursor and event. + // A deferred transaction cannot upgrade a stale WAL snapshot after + // a peer advances the same consumer and commits. + if _, err := q.Exec(` + UPDATE attachment_change_consumers + SET last_sequence = last_sequence WHERE consumer_key = ?`, consumerKey); err != nil { + return fmt.Errorf("lock attachment change consumer: %w", err) + } + } var current int64 var complete bool if err := q.QueryRow(` @@ -250,6 +260,17 @@ func (s *Store) AdvanceAttachmentChangeConsumer( return fmt.Errorf("verify attachment change acknowledgement: %w", err) } if !exists { + // A peer can advance the cursor and prune this event after our + // first cursor read. Treat that committed advance as success. + var advanced int64 + if err := q.QueryRow(` + SELECT last_sequence FROM attachment_change_consumers + WHERE consumer_key = ?`, consumerKey).Scan(&advanced); err != nil { + return fmt.Errorf("recheck attachment consumer cursor after pruned event: %w", err) + } + if advanced >= sequence { + return pruneAttachmentChanges(q) + } return errors.New("attachment change acknowledgement is not a retained event") } result, err := q.Exec(` diff --git a/internal/store/attachment_changes_test.go b/internal/store/attachment_changes_test.go index 8859fce10..c673a5c71 100644 --- a/internal/store/attachment_changes_test.go +++ b/internal/store/attachment_changes_test.go @@ -1,8 +1,10 @@ package store_test import ( + "context" "strings" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -119,6 +121,66 @@ func TestAttachmentChangeConsumersPruneOnlySharedConsumedPrefix(t *testing.T) { assert.Zero(attachmentChangeCount(t, f), "capture stops after the final consumer unregisters") } +func TestSQLiteAttachmentChangeAdvanceWaitsForWriterSlot(t *testing.T) { + require := require.New(t) + f := storetest.New(t) + if f.Store.IsPostgreSQL() { + t.Skip("SQLite writer-slot regression") + } + consumer, _, err := f.Store.RegisterAttachmentChangeConsumer(t.Context(), "document-index/v1") + require.NoError(err) + require.NoError(f.Store.CompleteAttachmentChangeReconciliation( + t.Context(), consumer.ConsumerKey, consumer.BaselineSequence, + )) + messageID := f.CreateMessage("attachment-consumer-writer-slot") + createJournalAttachment(t, f, messageID, "part:writer-slot", "e") + changes, err := f.Store.ListAttachmentChanges(t.Context(), consumer.ConsumerKey, 1) + require.NoError(err) + require.Len(changes, 1) + + holder, err := f.Store.DB().Conn(t.Context()) + require.NoError(err) + held := true + t.Cleanup(func() { + if held { + _, _ = holder.ExecContext(context.Background(), "ROLLBACK") + } + _ = holder.Close() + }) + _, err = holder.ExecContext(t.Context(), "BEGIN IMMEDIATE") + require.NoError(err) + _, err = holder.ExecContext(t.Context(), ` + INSERT INTO archive_metadata (key, value) + VALUES ('test.attachment-writer-slot', 'held')`) + require.NoError(err) + + result := make(chan error, 1) + go func() { + result <- f.Store.AdvanceAttachmentChangeConsumer( + t.Context(), consumer.ConsumerKey, changes[0].Sequence, + ) + }() + require.Eventually(func() bool { + return f.Store.DB().Stats().InUse >= 2 || len(result) > 0 + }, time.Second, time.Millisecond) + select { + case advanceErr := <-result: + require.NoError(advanceErr, "cursor advance returned while the SQLite writer slot was held") + default: + } + require.GreaterOrEqual(f.Store.DB().Stats().InUse, 2) + + select { + case advanceErr := <-result: + require.NoError(advanceErr, "cursor advance returned while the SQLite writer slot was held") + case <-time.After(50 * time.Millisecond): + } + _, err = holder.ExecContext(t.Context(), "COMMIT") + require.NoError(err) + held = false + require.NoError(<-result) +} + func TestAttachmentChangeJournalCapturesCascadeDeletion(t *testing.T) { require := require.New(t) f := storetest.New(t) diff --git a/internal/store/db_logger.go b/internal/store/db_logger.go index 1375ae8b8..8539781fe 100644 --- a/internal/store/db_logger.go +++ b/internal/store/db_logger.go @@ -314,11 +314,13 @@ func (t *loggedTx) QueryRowContext( type loggedRows struct { *sql.Rows - query string - args []any - reqID string - start time.Time - finalized bool + query string + args []any + reqID string + start time.Time + finalized bool + rowsErr error + rowsErrSet bool } // Next delegates to the embedded *sql.Rows but, on the first @@ -333,10 +335,21 @@ func (r *loggedRows) Next() bool { if r.Rows.Next() { return true } - r.finalize(nil) + r.rowsErr = r.Rows.Err() + r.rowsErrSet = true + r.finalize(r.rowsErr) return false } +// Err returns the row error captured when iteration finished. Keeping that +// result avoids a second Rows.Err call racing the context-close goroutine. +func (r *loggedRows) Err() error { + if r.rowsErrSet { + return r.rowsErr + } + return r.Rows.Err() +} + // Close finalizes timing for the early-exit path (caller broke // out of the Next loop) and always closes the underlying Rows. // Repeated calls remain safe: finalize is idempotent and @@ -344,25 +357,26 @@ func (r *loggedRows) Next() bool { // times. func (r *loggedRows) Close() error { err := r.Rows.Close() - r.finalize(err) + if !r.rowsErrSet { + r.rowsErr = r.Rows.Err() + r.rowsErrSet = true + } + logErr := err + if logErr == nil { + logErr = r.rowsErr + } + r.finalize(logErr) return err } -// finalize emits the timing log line exactly once. closeErr is -// the error returned by Rows.Close on the explicit-Close path -// and nil on the end-of-scan path; either way, when no close -// error is present we still consult Rows.Err() so iteration -// failures (context cancellation, driver scan errors) get -// logged as "sql error" instead of as a successful query. -func (r *loggedRows) finalize(closeErr error) { +// finalize emits the timing log line exactly once. logErr comes from +// Rows.Close or Rows.Err so iteration failures are logged as "sql error" +// instead of as a successful query. +func (r *loggedRows) finalize(logErr error) { if r.finalized { return } r.finalized = true - logErr := closeErr - if logErr == nil { - logErr = r.Err() - } logStmtWith("query", r.reqID, r.query, r.args, logErr, time.Since(r.start)) } diff --git a/internal/store/dialect_pg.go b/internal/store/dialect_pg.go index 78282ecb5..f045e8bbf 100644 --- a/internal/store/dialect_pg.go +++ b/internal/store/dialect_pg.go @@ -1876,8 +1876,7 @@ func (d *PostgreSQLDialect) MaintenanceTimeoutResetSQL() string { // isPgError checks if err is a pgconn.PgError with the given SQLSTATE code. func isPgError(err error, code string) bool { - var pgErr *pgconn.PgError - if errors.As(err, &pgErr) { + if pgErr, ok := errors.AsType[*pgconn.PgError](err); ok { return pgErr.Code == code } return false diff --git a/internal/store/dialect_sqlite.go b/internal/store/dialect_sqlite.go index 6f000eb32..9c3a2ce0f 100644 --- a/internal/store/dialect_sqlite.go +++ b/internal/store/dialect_sqlite.go @@ -1750,8 +1750,7 @@ func (d *SQLiteDialect) IsBusyError(err error) bool { if err == nil { return false } - var serr sqlite3.Error - if errors.As(err, &serr) { + if serr, ok := errors.AsType[sqlite3.Error](err); ok { return serr.Code == sqlite3.ErrBusy || serr.Code == sqlite3.ErrLocked } var serrPtr *sqlite3.Error diff --git a/internal/store/document_index.go b/internal/store/document_index.go index b59f4009d..30522b406 100644 --- a/internal/store/document_index.go +++ b/internal/store/document_index.go @@ -302,56 +302,98 @@ func (s *Store) ReconcileDocumentOccurrence( if attachmentID <= 0 || sourceSequence < 0 { return DocumentOccurrence{}, false, errors.New("document occurrence reconciliation has invalid coordinates") } - file, err := s.GetFileMetadata(ctx, attachmentID) + var occurrence DocumentOccurrence + var eligible bool + err := s.withTxContext(ctx, func(tx *loggedTx) error { + if err := s.lockDocumentOccurrenceAttachmentTx(ctx, tx, attachmentID); err != nil { + return err + } + file, found, err := s.getDocumentFileMetadataTx(ctx, tx, attachmentID) + if err != nil { + return err + } + if !found || !eligibleDocumentFile(file) { + return removeDocumentOccurrenceTx(tx, attachmentID, sourceSequence) + } + occurrence = DocumentOccurrence{ + OccurrenceKey: documentOccurrenceKey(file.MessageID, file.SourcePartKey, file.ID), + AttachmentID: file.ID, + MessageID: file.MessageID, + SourceID: file.SourceID, + SourcePartKey: file.SourcePartKey, + StableSourcePart: file.SourcePartKey != "", + CanonicalBlobHash: file.ContentHash, + Filename: file.Filename, + MIMEType: file.MimeType, + AttachmentRole: file.AttachmentRole, + RoleSource: file.RoleSource, + SourceSequence: sourceSequence, + } + if err := upsertDocumentOccurrenceTx(tx, occurrence); err != nil { + return err + } + eligible = true + return nil + }) if err != nil { return DocumentOccurrence{}, false, err } - live := false - if file != nil { - live, err = s.documentAttachmentIsLive(ctx, attachmentID) - if err != nil { - return DocumentOccurrence{}, false, err - } - } - if file == nil || !live || !eligibleDocumentFile(*file) { - if err := s.removeDocumentOccurrence(ctx, attachmentID, sourceSequence); err != nil { - return DocumentOccurrence{}, false, err - } - return DocumentOccurrence{}, false, nil - } - stable := file.SourcePartKey != "" - occurrence := DocumentOccurrence{ - OccurrenceKey: documentOccurrenceKey(file.MessageID, file.SourcePartKey, file.ID), - AttachmentID: file.ID, - MessageID: file.MessageID, - SourceID: file.SourceID, - SourcePartKey: file.SourcePartKey, - StableSourcePart: stable, - CanonicalBlobHash: file.ContentHash, - Filename: file.Filename, - MIMEType: file.MimeType, - AttachmentRole: file.AttachmentRole, - RoleSource: file.RoleSource, - SourceSequence: sourceSequence, - } - if err := s.upsertDocumentOccurrence(ctx, occurrence); err != nil { - return DocumentOccurrence{}, false, err + return occurrence, eligible, nil +} + +func (s *Store) lockDocumentOccurrenceAttachmentTx( + ctx context.Context, tx *loggedTx, attachmentID int64, +) error { + if s.IsPostgreSQL() { + // The occurrence row may not exist yet, so there is no row to lock. + // Serialize the eligibility read and its write by attachment identity. + if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock( + hashtextextended(CAST(? AS TEXT), 0))`, + fmt.Sprintf("msgvault.document_occurrence.attachment:%d", attachmentID), + ); err != nil { + return fmt.Errorf("lock document occurrence attachment: %w", err) + } + return nil + } + // Reserve SQLite's writer slot before reading attachment authority. A + // deferred transaction cannot upgrade a stale WAL snapshot after a + // concurrent reconciliation commits. + if _, err := tx.ExecContext(ctx, ` + UPDATE document_index_state SET revision = revision WHERE singleton = 1`); err != nil { + return fmt.Errorf("lock document occurrence reconciliation: %w", err) } - return occurrence, true, nil + return nil } -func (s *Store) documentAttachmentIsLive(ctx context.Context, attachmentID int64) (bool, error) { - var exists bool - err := s.db.QueryRowContext(ctx, s.dialect.Rebind(` - SELECT EXISTS ( - SELECT 1 FROM attachments a - JOIN messages m ON m.id = a.message_id - WHERE a.id = ? AND `+LiveMessagesWhere("m", true)+` - )`), attachmentID).Scan(&exists) +func (s *Store) getDocumentFileMetadataTx( + ctx context.Context, tx *loggedTx, attachmentID int64, +) (FileMetadata, bool, error) { + var file FileMetadata + err := tx.QueryRowContext(ctx, ` + SELECT a.id, a.message_id, m.conversation_id, + m.source_id, COALESCE(m.source_message_id, ''), + COALESCE(m.message_type, ''), COALESCE(c.conversation_type, ''), + COALESCE(a.filename, ''), COALESCE(a.mime_type, ''), COALESCE(a.size, 0), + COALESCE(a.content_hash, ''), COALESCE(a.storage_path, ''), + a.attachment_role, a.role_source, + COALESCE(a.source_part_key, ''), COALESCE(a.content_id, '') + FROM attachments a + JOIN messages m ON m.id = a.message_id + JOIN conversations c ON c.id = m.conversation_id + WHERE a.id = ? AND `+LiveMessagesWhere("m", true), attachmentID).Scan( + &file.ID, &file.MessageID, &file.ConversationID, + &file.SourceID, &file.SourceMessageID, &file.MessageType, &file.ConversationType, + &file.Filename, &file.MimeType, &file.Size, &file.ContentHash, &file.StoragePath, + &file.AttachmentRole, &file.RoleSource, &file.SourcePartKey, &file.ContentID, + ) + if errors.Is(err, sql.ErrNoRows) { + return FileMetadata{}, false, nil + } if err != nil { - return false, fmt.Errorf("check document attachment live state: %w", err) + return FileMetadata{}, false, fmt.Errorf("read document occurrence attachment metadata: %w", err) } - return exists, nil + normalizeFileMetadataStorage(&file) + return file, true, nil } func (s *Store) GetDocumentIndexRevision(ctx context.Context) (int64, error) { @@ -1191,69 +1233,68 @@ func documentOccurrenceMediaScopeSQL( return strings.Join(conditions, " AND "), args, nil } -func (s *Store) upsertDocumentOccurrence(ctx context.Context, occurrence DocumentOccurrence) error { - return s.withTxContext(ctx, func(tx *loggedTx) error { - var existing DocumentOccurrence - err := tx.QueryRow(` +func upsertDocumentOccurrenceTx(tx *loggedTx, occurrence DocumentOccurrence) error { + var existing DocumentOccurrence + err := tx.QueryRow(` SELECT occurrence_key, attachment_id, message_id, source_id, COALESCE(source_part_key, ''), stable_source_part, canonical_blob_hash, COALESCE(filename, ''), COALESCE(mime_type, ''), attachment_role, role_source, source_sequence FROM document_occurrences WHERE occurrence_key = ?`, occurrence.OccurrenceKey).Scan( - &existing.OccurrenceKey, &existing.AttachmentID, &existing.MessageID, - &existing.SourceID, &existing.SourcePartKey, &existing.StableSourcePart, - &existing.CanonicalBlobHash, &existing.Filename, &existing.MIMEType, - &existing.AttachmentRole, &existing.RoleSource, &existing.SourceSequence, - ) - if err == nil { - existingSequence := existing.SourceSequence - existing.SourceSequence = occurrence.SourceSequence - if existing == occurrence { - if existingSequence >= occurrence.SourceSequence { - return nil - } - if _, err := tx.Exec(` + &existing.OccurrenceKey, &existing.AttachmentID, &existing.MessageID, + &existing.SourceID, &existing.SourcePartKey, &existing.StableSourcePart, + &existing.CanonicalBlobHash, &existing.Filename, &existing.MIMEType, + &existing.AttachmentRole, &existing.RoleSource, &existing.SourceSequence, + ) + if err == nil { + existingSequence := existing.SourceSequence + existing.SourceSequence = occurrence.SourceSequence + if existing == occurrence { + if existingSequence >= occurrence.SourceSequence { + return nil + } + if _, err := tx.Exec(` UPDATE document_occurrences SET source_sequence = ?, reconciled_at = CURRENT_TIMESTAMP WHERE occurrence_key = ? AND source_sequence = ?`, - occurrence.SourceSequence, occurrence.OccurrenceKey, existingSequence, - ); err != nil { - return fmt.Errorf("advance document occurrence source sequence: %w", err) - } - return nil + occurrence.SourceSequence, occurrence.OccurrenceKey, existingSequence, + ); err != nil { + return fmt.Errorf("advance document occurrence source sequence: %w", err) } - existing.SourceSequence = existingSequence - } - if err == nil && existing.SourceSequence > occurrence.SourceSequence { return nil } - if err != nil && !errors.Is(err, sql.ErrNoRows) { - return fmt.Errorf("read document occurrence: %w", err) - } - var attachmentSequence int64 - err = tx.QueryRow(` + existing.SourceSequence = existingSequence + } + if err == nil && existing.SourceSequence > occurrence.SourceSequence { + return nil + } + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("read document occurrence: %w", err) + } + var attachmentSequence int64 + err = tx.QueryRow(` SELECT source_sequence FROM document_occurrences WHERE attachment_id = ?`, - occurrence.AttachmentID, - ).Scan(&attachmentSequence) - if err == nil && attachmentSequence > occurrence.SourceSequence { - return nil - } - if err != nil && !errors.Is(err, sql.ErrNoRows) { - return fmt.Errorf("read document occurrence attachment sequence: %w", err) - } - removedResult, err := tx.Exec(` + occurrence.AttachmentID, + ).Scan(&attachmentSequence) + if err == nil && attachmentSequence > occurrence.SourceSequence { + return nil + } + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return fmt.Errorf("read document occurrence attachment sequence: %w", err) + } + removedResult, err := tx.Exec(` DELETE FROM document_occurrences WHERE attachment_id = ? AND occurrence_key != ? AND source_sequence <= ?`, - occurrence.AttachmentID, occurrence.OccurrenceKey, occurrence.SourceSequence, - ) - if err != nil { - return fmt.Errorf("remove replaced document occurrence: %w", err) - } - removed, err := removedResult.RowsAffected() - if err != nil { - return fmt.Errorf("read replaced document occurrence count: %w", err) - } - result, err := tx.Exec(` + occurrence.AttachmentID, occurrence.OccurrenceKey, occurrence.SourceSequence, + ) + if err != nil { + return fmt.Errorf("remove replaced document occurrence: %w", err) + } + removed, err := removedResult.RowsAffected() + if err != nil { + return fmt.Errorf("read replaced document occurrence count: %w", err) + } + result, err := tx.Exec(` INSERT INTO document_occurrences (occurrence_key, attachment_id, message_id, source_id, source_part_key, stable_source_part, canonical_blob_hash, @@ -1273,47 +1314,44 @@ func (s *Store) upsertDocumentOccurrence(ctx context.Context, occurrence Documen source_sequence = EXCLUDED.source_sequence, reconciled_at = CURRENT_TIMESTAMP WHERE document_occurrences.source_sequence <= EXCLUDED.source_sequence`, - occurrence.OccurrenceKey, occurrence.AttachmentID, occurrence.MessageID, - occurrence.SourceID, nullIfEmpty(occurrence.SourcePartKey), - occurrence.StableSourcePart, occurrence.CanonicalBlobHash, - nullIfEmpty(occurrence.Filename), nullIfEmpty(occurrence.MIMEType), - occurrence.AttachmentRole, occurrence.RoleSource, occurrence.SourceSequence, - ) - if err != nil { - return fmt.Errorf("upsert document occurrence: %w", err) - } - changed, err := result.RowsAffected() - if err != nil { - return fmt.Errorf("read document occurrence upsert count: %w", err) - } - if changed == 0 { - return nil - } - if removed > 0 { - // The delete trigger already advanced the revision for this atomic - // replacement. One invalidation is sufficient for the new row too. - return nil - } - return bumpDocumentIndexRevision(tx) - }) + occurrence.OccurrenceKey, occurrence.AttachmentID, occurrence.MessageID, + occurrence.SourceID, nullIfEmpty(occurrence.SourcePartKey), + occurrence.StableSourcePart, occurrence.CanonicalBlobHash, + nullIfEmpty(occurrence.Filename), nullIfEmpty(occurrence.MIMEType), + occurrence.AttachmentRole, occurrence.RoleSource, occurrence.SourceSequence, + ) + if err != nil { + return fmt.Errorf("upsert document occurrence: %w", err) + } + changed, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("read document occurrence upsert count: %w", err) + } + if changed == 0 { + return nil + } + if removed > 0 { + // The delete trigger already advanced the revision for this atomic + // replacement. One invalidation is sufficient for the new row too. + return nil + } + return bumpDocumentIndexRevision(tx) } -func (s *Store) removeDocumentOccurrence(ctx context.Context, attachmentID, sourceSequence int64) error { - return s.withTxContext(ctx, func(tx *loggedTx) error { - result, err := tx.Exec(` +func removeDocumentOccurrenceTx(tx *loggedTx, attachmentID, sourceSequence int64) error { + result, err := tx.Exec(` DELETE FROM document_occurrences WHERE attachment_id = ? AND source_sequence <= ?`, attachmentID, sourceSequence) - if err != nil { - return fmt.Errorf("remove document occurrence: %w", err) - } - _, err = result.RowsAffected() - if err != nil { - return fmt.Errorf("read removed document occurrence count: %w", err) - } - // The database trigger advances the revision when a row is removed, - // including when foreign-key cascades bypass this method. - return nil - }) + if err != nil { + return fmt.Errorf("remove document occurrence: %w", err) + } + _, err = result.RowsAffected() + if err != nil { + return fmt.Errorf("read removed document occurrence count: %w", err) + } + // The database trigger advances the revision when a row is removed, + // including when foreign-key cascades bypass this method. + return nil } func bumpDocumentIndexRevision(q querier) error { diff --git a/internal/store/document_search_internal_test.go b/internal/store/document_search_internal_test.go index e35805423..3ace852d9 100644 --- a/internal/store/document_search_internal_test.go +++ b/internal/store/document_search_internal_test.go @@ -9,12 +9,12 @@ import ( func TestFuseDocumentSearchRowsCapsCombinedSignals(t *testing.T) { contentRows := []documentSearchRow{ - {DocumentSearchResult: DocumentSearchResult{OccurrenceKey: "content-1", AttachmentID: 1}, ContentRank: 1}, - {DocumentSearchResult: DocumentSearchResult{OccurrenceKey: "content-2", AttachmentID: 2}, ContentRank: 2}, + {OccurrenceKey: "content-1", AttachmentID: 1, ContentRank: 1}, + {OccurrenceKey: "content-2", AttachmentID: 2, ContentRank: 2}, } filenameRows := []documentSearchRow{ - {DocumentSearchResult: DocumentSearchResult{OccurrenceKey: "filename-1", AttachmentID: 3}, FilenameRank: 1}, - {DocumentSearchResult: DocumentSearchResult{OccurrenceKey: "filename-2", AttachmentID: 4}, FilenameRank: 2}, + {OccurrenceKey: "filename-1", AttachmentID: 3, FilenameRank: 1}, + {OccurrenceKey: "filename-2", AttachmentID: 4, FilenameRank: 2}, } rows, truncated := fuseDocumentSearchRows(contentRows, filenameRows, nil, 3) diff --git a/internal/store/files.go b/internal/store/files.go index 4e504aa5b..1d50cca37 100644 --- a/internal/store/files.go +++ b/internal/store/files.go @@ -97,20 +97,7 @@ func (s *Store) GetFileMetadataBatch(ctx context.Context, ids []int64) (map[int6 &file.AttachmentRole, &file.RoleSource, &file.SourcePartKey, &file.ContentID); err != nil { return nil, fmt.Errorf("scan file metadata: %w", err) } - lowerPath := strings.ToLower(file.StoragePath) - switch { - case strings.HasPrefix(lowerPath, "http://") || strings.HasPrefix(lowerPath, "https://"): - file.URL = file.StoragePath - file.StoragePath = "" - file.ContentHash = "" - case file.ContentHash == "": - // Duplicate-content aliases (see normalizeDiscordAttachmentRefs) - // keep a trusted CAS path with an empty hash. Recover the hash so - // the file endpoints classify the alias as locally available. - if pathHash, ok := casPathHash(file.StoragePath); ok { - file.ContentHash = pathHash - } - } + normalizeFileMetadataStorage(&file) result[file.ID] = file } if err := rows.Err(); err != nil { @@ -118,3 +105,20 @@ func (s *Store) GetFileMetadataBatch(ctx context.Context, ids []int64) (map[int6 } return result, nil } + +func normalizeFileMetadataStorage(file *FileMetadata) { + lowerPath := strings.ToLower(file.StoragePath) + switch { + case strings.HasPrefix(lowerPath, "http://") || strings.HasPrefix(lowerPath, "https://"): + file.URL = file.StoragePath + file.StoragePath = "" + file.ContentHash = "" + case file.ContentHash == "": + // Duplicate-content aliases (see normalizeDiscordAttachmentRefs) + // keep a trusted CAS path with an empty hash. Recover the hash so + // the file endpoints classify the alias as locally available. + if pathHash, ok := casPathHash(file.StoragePath); ok { + file.ContentHash = pathHash + } + } +} diff --git a/internal/store/sqlite_error_test.go b/internal/store/sqlite_error_test.go index 906f70480..a5f2ef5ef 100644 --- a/internal/store/sqlite_error_test.go +++ b/internal/store/sqlite_error_test.go @@ -34,14 +34,11 @@ func TestIsSQLiteError_PointerForm(t *testing.T) { ExtendedCode: sqlite3.ErrConstraintForeignKey, } - // Wrap the error - wrappedErr := fmt.Errorf("insert failed: %w", sqliteErr) - // sqlite3.Error.Error() returns the code description, e.g. "constraint failed" - assert.True(t, isSQLiteError(wrappedErr, "constraint failed"), + assert.True(t, isSQLiteError(sqliteErr, "constraint failed"), "isSQLiteError should match constraint error via pointer, got: %v", sqliteErr.Error()) - assert.False(t, isSQLiteError(wrappedErr, "no such table"), + assert.False(t, isSQLiteError(sqliteErr, "no such table"), "isSQLiteError should not match unrelated substring via pointer") } diff --git a/internal/store/store.go b/internal/store/store.go index fe8566955..f2395a74b 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -85,8 +85,7 @@ const defaultSQLiteParams = "?_journal_mode=WAL&_busy_timeout=30000&_synchronous // SQLiteDialect's error predicates are thin wrappers around this helper; it also // services subset.go (which has not been migrated to Dialect). func isSQLiteError(err error, substr string) bool { - var sqliteErr sqlite3.Error - if errors.As(err, &sqliteErr) { + if sqliteErr, ok := errors.AsType[sqlite3.Error](err); ok { return strings.Contains(sqliteErr.Error(), substr) } var sqliteErrPtr *sqlite3.Error diff --git a/internal/sync/incremental.go b/internal/sync/incremental.go index 6884f32e2..6321d610d 100644 --- a/internal/sync/incremental.go +++ b/internal/sync/incremental.go @@ -107,8 +107,7 @@ func (s *Syncer) Incremental(ctx context.Context, source *store.Source) (summary historyResp, err := s.client.ListHistory(ctx, startHistoryID, pageToken) if err != nil { // Check for 404 - history too old - var notFound *gmail.NotFoundError - if errors.As(err, ¬Found) { + if _, ok := errors.AsType[*gmail.NotFoundError](err); ok { s.logger.Info("gmail history expired; full sync required") _ = s.store.FailSync(syncID, "history too old") // Callers fall back to a full sync on ErrHistoryExpired. @@ -390,8 +389,7 @@ func (s *Syncer) handleLabelChange(ctx context.Context, syncID, sourceID int64, // to a debug-level message since deleted messages are expected during // incremental sync (e.g., spam auto-deleted between sync runs). func (s *Syncer) logLabelChangeError(action, messageID string, err error) { - var notFound *gmail.NotFoundError - if errors.As(err, ¬Found) { + if _, ok := errors.AsType[*gmail.NotFoundError](err); ok { s.logger.Debug("skipping label "+action+": message deleted from Gmail", "id", messageID) } else { s.logger.Warn("failed to handle label "+action, "id", messageID, "error", err) diff --git a/internal/syncerr/transient.go b/internal/syncerr/transient.go index e90c2e2e2..1d7a296f3 100644 --- a/internal/syncerr/transient.go +++ b/internal/syncerr/transient.go @@ -16,12 +16,10 @@ func IsTransientNetwork(err error) bool { if err == nil { return false } - var dnsErr *net.DNSError - if errors.As(err, &dnsErr) { + if _, ok := errors.AsType[*net.DNSError](err); ok { return true } - var opErr *net.OpError - if errors.As(err, &opErr) { + if _, ok := errors.AsType[*net.OpError](err); ok { return true } var netErr net.Error diff --git a/internal/taskclient/client.go b/internal/taskclient/client.go index 335fee8bf..3094f4f97 100644 --- a/internal/taskclient/client.go +++ b/internal/taskclient/client.go @@ -194,8 +194,7 @@ func (c *Client) HasAuthentication() bool { return c.apiKey != "" || c.endpoi func (c *Client) Capabilities(ctx context.Context) (Capabilities, error) { var result Capabilities if err := c.doJSON(ctx, http.MethodGet, "/api/v1/capabilities", nil, nil, &result, http.StatusOK); err != nil { - var statusErr *httpStatusError - if errors.As(err, &statusErr) { + if statusErr, ok := errors.AsType[*httpStatusError](err); ok { return Capabilities{}, &httpStatusError{ statusCode: statusErr.statusCode, classification: classifyCapabilityHTTPStatus(statusErr.statusCode), diff --git a/internal/tui/model.go b/internal/tui/model.go index 7bcf63c20..82358cc7d 100644 --- a/internal/tui/model.go +++ b/internal/tui/model.go @@ -286,15 +286,13 @@ func New(engine query.Engine, opts Options) Model { analyticsNotice: opts.AnalyticsNotice, aggregateLimit: aggLimit, threadMessageLimit: threadLimit, - viewState: viewState{ - level: levelAggregates, - viewType: query.ViewSenders, - timeGranularity: query.TimeMonth, - sortField: query.SortByCount, - sortDirection: query.SortDesc, - msgSortField: query.MessageSortByDate, - msgSortDirection: query.SortDesc, - }, + level: levelAggregates, + viewType: query.ViewSenders, + timeGranularity: query.TimeMonth, + sortField: query.SortByCount, + sortDirection: query.SortDesc, + msgSortField: query.MessageSortByDate, + msgSortDirection: query.SortDesc, meetingState: meetingState{ searchInput: meetingInput, detailSearchInput: meetingDetailInput, diff --git a/internal/vcard/model.go b/internal/vcard/model.go index 11d3509f2..dd4b2e58d 100644 --- a/internal/vcard/model.go +++ b/internal/vcard/model.go @@ -30,8 +30,6 @@ type Card struct { // blank logical lines, original line endings, or physical folding. The JSON // form is the persisted resource metadata shape; a RawValue that is not valid // UTF-8 (a CHARSET-declared legacy value) travels as raw_value_base64. -// -//nolint:recvcheck // encoding/json requires the pointer receiver for UnmarshalJSON type Property struct { Group string `json:"group,omitempty"` Name string `json:"name"` diff --git a/internal/vector/visual/voyage.go b/internal/vector/visual/voyage.go index be8e8133a..e7c17eb65 100644 --- a/internal/vector/visual/voyage.go +++ b/internal/vector/visual/voyage.go @@ -189,9 +189,8 @@ func mapVoyageError(ctx context.Context, err error) error { if ctx.Err() != nil && errors.Is(err, ctx.Err()) { return err } - var voyageErr *voyage.ProviderError statusCode := 0 - if errors.As(err, &voyageErr) { + if voyageErr, ok := errors.AsType[*voyage.ProviderError](err); ok { statusCode = voyageErr.StatusCode } switch { diff --git a/nix/package.nix b/nix/package.nix index 0923c67d7..dd7e87db9 100644 --- a/nix/package.nix +++ b/nix/package.nix @@ -16,7 +16,7 @@ buildGoModule { src = gitignoreSource ../.; - vendorHash = "sha256-CR32j6Xmc/jbqQqlshwVSWULDJTyq2Opb7hEtO6q8Q0="; + vendorHash = "sha256-mywedDZlx89nFjHuZmSqRxLnPMzVVNR4dqM2RwGWYbI="; proxyVendor = true; # Bun's copyfile backend can install incomplete packages when fetchBunDeps' diff --git a/scripts/docs_assets_test.go b/scripts/docs_assets_test.go index 5f74a1ee2..ff242f88e 100644 --- a/scripts/docs_assets_test.go +++ b/scripts/docs_assets_test.go @@ -388,9 +388,9 @@ func TestDocsScreenshotDockerfileGoVersionMatchesModule(t *testing.T) { dockerfile, err := os.ReadFile(filepath.Join("..", "docs", "screenshots", "Dockerfile")) require.NoError(err) - moduleGoVersion := regexp.MustCompile(`(?m)^go\s+([0-9]+\.[0-9]+)`).FindStringSubmatch(string(goMod)) + moduleGoVersion := regexp.MustCompile(`(?m)^go\s+([0-9]+\.[0-9]+\.[0-9]+)`).FindStringSubmatch(string(goMod)) require.Len(moduleGoVersion, 2) - dockerGoVersion := regexp.MustCompile(`(?m)^FROM\s+golang:([0-9]+\.[0-9]+)-`).FindStringSubmatch(string(dockerfile)) + dockerGoVersion := regexp.MustCompile(`(?m)^FROM\s+golang:([0-9]+\.[0-9]+\.[0-9]+)-`).FindStringSubmatch(string(dockerfile)) require.Len(dockerGoVersion, 2) assert.Equal(t, moduleGoVersion[1], dockerGoVersion[1])