From 399a7ea9a0170557c932ee54daf18c0f2e4c1ac2 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 22 Apr 2026 21:40:58 -0400 Subject: [PATCH 1/6] =?UTF-8?q?feat:=20v0.2.0=20=E2=80=94=20Atlas=20driver?= =?UTF-8?q?,=20lint=20tool,=20atlas=20plugin=20wiring,=20Dockerfile?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task 27: Atlas driver (ariga.io/atlas v1.2.0) - internal/atlas/driver.go: full MigrationDriver implementation (Up/Down/Status/Goto) - internal/atlas/revisions.go: SQL-backed RevisionReadWriter (atlas_schema_revisions table) - internal/atlas/driver_test.go: Name + UpDownStatus integration tests - Auto-generates atlas.sum when absent so Atlas CLI is not required for basic ops - Down reads paired .down.sql files for rollback SQL Task 28: Lint tool - pkg/lint/lint.go: static analysis (L001 ordering gap, L002 duplicate, L003 naming, L004 dangerous ops, L005 missing .down.sql, L006 empty file) - pkg/lint/lint_test.go: one test per lint code - pkg/cli/lint.go: `migrate lint ` cobra subcommand with --json output Task 27 (cont): Atlas plugin binary wiring - internal/atlasplugin/plugin.go: replace stub with real CreateModule dispatcher - internal/atlasplugin/modules.go: atlasMigrationsModule, driverModule, driverBackedModule Task 29: workflow-migrate standalone binary - cmd/workflow-migrate/Dockerfile: distroless/static OCI image for pre-deploy jobs Task 30: wfctl CLI update - pkg/cli/root.go: add lint subcommand + atlas driver support - plugin.json: update description, keywords, migrationDrivers, cliCommands Co-Authored-By: Claude Sonnet 4.6 --- cmd/workflow-migrate/Dockerfile | 14 ++ go.mod | 11 + go.sum | 24 +++ internal/atlas/driver.go | 358 ++++++++++++++++++++++++++++++++ internal/atlas/driver_test.go | 125 +++++++++++ internal/atlas/revisions.go | 164 +++++++++++++++ internal/atlasplugin/modules.go | 149 +++++++++++++ internal/atlasplugin/plugin.go | 49 ++++- pkg/cli/lint.go | 67 ++++++ pkg/cli/root.go | 10 +- pkg/lint/lint.go | 228 ++++++++++++++++++++ pkg/lint/lint_test.go | 121 +++++++++++ plugin.json | 9 +- 13 files changed, 1312 insertions(+), 17 deletions(-) create mode 100644 cmd/workflow-migrate/Dockerfile create mode 100644 internal/atlas/driver.go create mode 100644 internal/atlas/driver_test.go create mode 100644 internal/atlas/revisions.go create mode 100644 internal/atlasplugin/modules.go create mode 100644 pkg/cli/lint.go create mode 100644 pkg/lint/lint.go create mode 100644 pkg/lint/lint_test.go diff --git a/cmd/workflow-migrate/Dockerfile b/cmd/workflow-migrate/Dockerfile new file mode 100644 index 0000000..3b022a2 --- /dev/null +++ b/cmd/workflow-migrate/Dockerfile @@ -0,0 +1,14 @@ +# syntax=docker/dockerfile:1 +# Build stage +FROM golang:1.26-alpine AS builder +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY . . +RUN CGO_ENABLED=0 GOARCH=amd64 GOOS=linux \ + go build -ldflags="-s -w" -o /workflow-migrate ./cmd/workflow-migrate + +# Runtime stage — distroless/static has no shell, minimal attack surface +FROM gcr.io/distroless/static:nonroot +COPY --from=builder /workflow-migrate /workflow-migrate +ENTRYPOINT ["/workflow-migrate"] diff --git a/go.mod b/go.mod index e9b47ea..9905d85 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/GoCodeAlone/workflow-plugin-migrations go 1.26.0 require ( + ariga.io/atlas v1.2.0 github.com/GoCodeAlone/workflow v0.18.2 github.com/fergusstrange/embedded-postgres v1.34.0 github.com/golang-migrate/migrate/v4 v4.19.1 @@ -35,6 +36,9 @@ require ( github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.55.0 // indirect github.com/IBM/sarama v1.47.0 // indirect github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/agext/levenshtein v1.2.1 // indirect + github.com/apparentlymart/go-textseg/v13 v13.0.0 // indirect + github.com/apparentlymart/go-textseg/v15 v15.0.0 // indirect github.com/aws/aws-sdk-go-v2 v1.41.5 // indirect github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 // indirect github.com/aws/aws-sdk-go-v2/config v1.32.12 // indirect @@ -63,6 +67,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/sts v1.41.9 // indirect github.com/aws/smithy-go v1.24.2 // indirect github.com/beorn7/perks v1.0.1 // indirect + github.com/bmatcuk/doublestar v1.3.4 // indirect github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect @@ -91,9 +96,11 @@ require ( github.com/go-jose/go-jose/v4 v4.1.4 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect + github.com/go-openapi/inflect v0.19.0 // indirect github.com/godbus/dbus/v5 v5.2.2 // indirect github.com/golang-jwt/jwt/v5 v5.3.1 // indirect github.com/golobby/cast v1.3.3 // indirect + github.com/google/go-cmp v0.7.0 // indirect github.com/google/go-querystring v1.2.0 // indirect github.com/google/s2a-go v0.1.9 // indirect github.com/google/uuid v1.6.0 // indirect @@ -112,6 +119,7 @@ require ( github.com/hashicorp/go-uuid v1.0.3 // indirect github.com/hashicorp/golang-lru v1.0.2 // indirect github.com/hashicorp/hcl v1.0.1-vault-7 // indirect + github.com/hashicorp/hcl/v2 v2.13.0 // indirect github.com/hashicorp/vault/api v1.23.0 // indirect github.com/hashicorp/yamux v0.1.2 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect @@ -134,6 +142,7 @@ require ( github.com/mfridman/interpolate v0.0.2 // indirect github.com/mitchellh/go-homedir v1.1.0 // indirect github.com/mitchellh/go-testing-interface v1.14.1 // indirect + github.com/mitchellh/go-wordwrap v1.0.1 // indirect github.com/mitchellh/mapstructure v1.5.0 // indirect github.com/moby/docker-image-spec v1.3.1 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect @@ -165,6 +174,8 @@ require ( github.com/xdg-go/stringprep v1.0.4 // indirect github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect github.com/zalando/go-keyring v0.2.8 // indirect + github.com/zclconf/go-cty v1.14.4 // indirect + github.com/zclconf/go-cty-yaml v1.1.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/detectors/gcp v1.43.0 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 // indirect diff --git a/go.sum b/go.sum index 9f35900..d263e08 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +ariga.io/atlas v1.2.0 h1:PJZzG1zTA8SzMR1giJ7XnQe4VjfDWEbJjHnIaK45BFg= +ariga.io/atlas v1.2.0/go.mod h1:vg7qWSatkNqs04Y4Lheg7vR4bGX0wy51Wz4FIMFVr3U= cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4= cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= @@ -36,6 +38,8 @@ github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 h1:XRzhVemXdgv github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0/go.mod h1:HKpQxkWaGLJ+D/5H8QRpyQXA1eKjxkFlOMwck5+33Jk= github.com/BurntSushi/toml v1.6.0 h1:dRaEfpa2VI55EwlIW72hMRHdWouJeRF7TPYhI+AUQjk= github.com/BurntSushi/toml v1.6.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= +github.com/DATA-DOG/go-sqlmock v1.5.0 h1:Shsta01QNfFxHCfpW6YH2STWB0MudeXXEWMr20OEh60= +github.com/DATA-DOG/go-sqlmock v1.5.0/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM= github.com/DataDog/datadog-go/v5 v5.8.3 h1:s58CUJ9s8lezjhTNJO/SxkPBv2qZjS3ktpRSqGF5n0s= github.com/DataDog/datadog-go/v5 v5.8.3/go.mod h1:K9kcYBlxkcPP8tvvjZZKs/m1edNAUFzBbdpTUKfCsuw= github.com/GoCodeAlone/go-plugin v1.7.0 h1:EwnhqPlXiNmp85S+MXnKKvm3YlfA6O4NzBb4+GSlEVY= @@ -63,10 +67,16 @@ github.com/IBM/sarama v1.47.0/go.mod h1:7gLLIU97nznOmA6TX++Qds+DRxH89P2XICY2KAQU github.com/Microsoft/go-winio v0.5.0/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/agext/levenshtein v1.2.1 h1:QmvMAjj2aEICytGiWzmxoE0x2KZvE0fvmqMOfy2tjT8= +github.com/agext/levenshtein v1.2.1/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558= github.com/alicebob/miniredis/v2 v2.36.1 h1:Dvc5oAnNOr7BIfPn7tF269U8DvRW1dBG2D5n0WrfYMI= github.com/alicebob/miniredis/v2 v2.36.1/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= github.com/antithesishq/antithesis-sdk-go v0.5.0-default-no-op h1:Ucf+QxEKMbPogRO5guBNe5cgd9uZgfoJLOYs8WWhtjM= github.com/antithesishq/antithesis-sdk-go v0.5.0-default-no-op/go.mod h1:IUpT2DPAKh6i/YhSbt6Gl3v2yvUZjmKncl7U91fup7E= +github.com/apparentlymart/go-textseg/v13 v13.0.0 h1:Y+KvPE1NYz0xl601PVImeQfFyEy6iT90AvPUL1NNfNw= +github.com/apparentlymart/go-textseg/v13 v13.0.0/go.mod h1:ZK2fH7c4NqDTLtiYLvIkEghdlcqw7yxLeM89kiTRPUo= +github.com/apparentlymart/go-textseg/v15 v15.0.0 h1:uYvfpb3DyLSCGWnctWKGj857c6ew1u1fNQOlOtuGxQY= +github.com/apparentlymart/go-textseg/v15 v15.0.0/go.mod h1:K8XmNZdhEBkdlyDdvbmmsvpAG721bKi0joRfFdHIWJ4= github.com/aws/aws-sdk-go-v2 v1.41.5 h1:dj5kopbwUsVUVFgO4Fi5BIT3t4WyqIDjGKCangnV/yY= github.com/aws/aws-sdk-go-v2 v1.41.5/go.mod h1:mwsPRE8ceUUpiTgF7QmQIJ7lgsKUPQOUl3o72QBrE1o= github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.8 h1:eBMB84YGghSocM7PsjmmPffTa+1FBUeNvGvFou6V/4o= @@ -123,6 +133,8 @@ github.com/aws/smithy-go v1.24.2 h1:FzA3bu/nt/vDvmnkg+R8Xl46gmzEDam6mZ1hzmwXFng= github.com/aws/smithy-go v1.24.2/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/bmatcuk/doublestar v1.3.4 h1:gPypJ5xD31uhX6Tf54sDPUOBXTqKH4c9aPY66CyQrS0= +github.com/bmatcuk/doublestar v1.3.4/go.mod h1:wiQtGV+rzVYxB7WIlirSN++5HPtPlXEo9MEoZQC/PmE= github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= @@ -207,6 +219,8 @@ github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/go-openapi/inflect v0.19.0 h1:9jCH9scKIbHeV9m12SmPilScz6krDxKRasNNSNPXu/4= +github.com/go-openapi/inflect v0.19.0/go.mod h1:lHpZVlpIQqLyKwJ4N+YSc9hchQy/i12fJykb83CRBH4= github.com/go-test/deep v1.1.1 h1:0r/53hagsehfO4bzD2Pgr/+RgHqhmf+k1Bpse2cTu1U= github.com/go-test/deep v1.1.1/go.mod h1:5C2ZWiW0ErCdrYzpqxLbTX7MG14M9iiw8DgHncVwcsE= github.com/godbus/dbus/v5 v5.2.2 h1:TUR3TgtSVDmjiXOgAAyaZbYmIeP3DPkld3jgKGV8mXQ= @@ -278,6 +292,8 @@ github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/hashicorp/hcl v1.0.1-vault-7 h1:ag5OxFVy3QYTFTJODRzTKVZ6xvdfLLCA1cy/Y6xGI0I= github.com/hashicorp/hcl v1.0.1-vault-7/go.mod h1:XYhtn6ijBSAj6n4YqAaf7RBPS4I06AItNorpy+MoQNM= +github.com/hashicorp/hcl/v2 v2.13.0 h1:0Apadu1w6M11dyGFxWnmhhcMjkbAiKCv7G1r/2QgCNc= +github.com/hashicorp/hcl/v2 v2.13.0/go.mod h1:e4z5nxYlWNPdDSNYX+ph14EvWYMFm3eP0zIUqPc2jr0= github.com/hashicorp/vault/api v1.23.0 h1:gXgluBsSECfRWTSW9niY2jwg2e9mMJc4WoHNv4g3h6A= github.com/hashicorp/vault/api v1.23.0/go.mod h1:zransKiB9ftp+kgY8ydjnvCU7Wk8i9L0DYWpXeMj9ko= github.com/hashicorp/yamux v0.1.2 h1:XtB8kyFOyHXYVFnwT5C3+Bdo8gArse7j2AQ0DA0Uey8= @@ -345,6 +361,8 @@ github.com/mitchellh/go-homedir v1.1.0 h1:lukF9ziXFxDFPkA1vsr5zpc1XuPDn/wFntq5mG github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0= github.com/mitchellh/go-testing-interface v1.14.1 h1:jrgshOhYAUVNMAJiKbEu7EqAwgJJ2JqpQmpLJOu07cU= github.com/mitchellh/go-testing-interface v1.14.1/go.mod h1:gfgS7OtZj6MA4U1UrDRp04twqAjfvlZyCfX3sDjEym8= +github.com/mitchellh/go-wordwrap v1.0.1 h1:TLuKupo69TCn6TQSyGxwI1EblZZEsQ0vMlAFQflz0v0= +github.com/mitchellh/go-wordwrap v1.0.1/go.mod h1:R62XHJLzvMFRBbcrT7m7WgmE1eOyTSsCt+hzestvNj0= github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY= github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= @@ -419,6 +437,8 @@ github.com/ryanuber/go-glob v1.0.0 h1:iQh3xXAumdQ+4Ufa5b25cRpC5TYKlno6hsv6Cb3pkB github.com/ryanuber/go-glob v1.0.0/go.mod h1:807d1WSdnB0XRJzKNil9Om6lcp/3a0v4qIHxIXzX/Yc= github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 h1:KRzFb2m7YtdldCEkzs6KqmJw4nqEVZGK7IN2kJkjTuQ= github.com/santhosh-tekuri/jsonschema/v6 v6.0.2/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU= +github.com/sergi/go-diff v1.0.0 h1:Kpca3qRNrduNnOQeazBd0ysaKrUJiIuISHxogkT9RPQ= +github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo= github.com/sethvargo/go-retry v0.3.0 h1:EEt31A35QhrcRZtrYFDTBg91cqZVnFL2navjDrah2SE= github.com/sethvargo/go-retry v0.3.0/go.mod h1:mNX17F0C/HguQMyMyJxcnU471gOZGxCLyYaFyAZraas= github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= @@ -462,6 +482,10 @@ github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= github.com/zalando/go-keyring v0.2.8 h1:6sD/Ucpl7jNq10rM2pgqTs0sZ9V3qMrqfIIy5YPccHs= github.com/zalando/go-keyring v0.2.8/go.mod h1:tsMo+VpRq5NGyKfxoBVjCuMrG47yj8cmakZDO5QGii0= +github.com/zclconf/go-cty v1.14.4 h1:uXXczd9QDGsgu0i/QFR/hzI5NYCHLf6NQw/atrbnhq8= +github.com/zclconf/go-cty v1.14.4/go.mod h1:VvMs5i0vgZdhYawQNq5kePSpLAoz8u1xvZgrPIxfnZE= +github.com/zclconf/go-cty-yaml v1.1.0 h1:nP+jp0qPHv2IhUVqmQSzjvqAWcObN0KBkUl2rWBdig0= +github.com/zclconf/go-cty-yaml v1.1.0/go.mod h1:9YLUH4g7lOhVWqUbctnVlZ5KLpg7JAprQNgxSZ1Gyxs= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= diff --git a/internal/atlas/driver.go b/internal/atlas/driver.go new file mode 100644 index 0000000..712ae2c --- /dev/null +++ b/internal/atlas/driver.go @@ -0,0 +1,358 @@ +// Package atlas provides a MigrationDriver backed by ariga.io/atlas v1. +// Atlas uses versioned SQL migration files with an atlas.sum integrity file. +// If atlas.sum is absent the driver auto-generates it so users don't need the +// Atlas CLI installed for basic up/down/status operations. +package atlas + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" + + atlmigrate "ariga.io/atlas/sql/migrate" + atlpg "ariga.io/atlas/sql/postgres" + + "github.com/GoCodeAlone/workflow/interfaces" + _ "github.com/jackc/pgx/v5/stdlib" +) + +// Driver implements interfaces.MigrationDriver using ariga.io/atlas. +type Driver struct{} + +// New returns a new Atlas Driver. +func New() *Driver { return &Driver{} } + +// Name returns the driver name used in plugin configuration. +func (d *Driver) Name() string { return "atlas" } + +// Up applies all pending migrations. +func (d *Driver) Up(ctx context.Context, req interfaces.MigrationRequest) (interfaces.MigrationResult, error) { + if err := req.Validate(); err != nil { + return interfaces.MigrationResult{}, err + } + start := time.Now() + + db, dir, rrw, drv, cleanup, err := open(req) + if err != nil { + return interfaces.MigrationResult{}, err + } + defer cleanup() + _ = db + + ex, err := atlmigrate.NewExecutor(drv, dir, rrw) + if err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas: executor: %w", err) + } + + if err := ex.ExecuteN(ctx, 0); err != nil && !errors.Is(err, atlmigrate.ErrNoPendingFiles) { + return interfaces.MigrationResult{}, fmt.Errorf("atlas up: %w", err) + } + + // Read applied revisions to report what changed. + revs, err := rrw.ReadRevisions(ctx) + if err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas up: read revisions: %w", err) + } + + // Collect versions applied in this run by comparing against what was applied + // before (we can't diff easily, so report all applied-without-error revisions). + var applied []string + for _, r := range revs { + if r.Error == "" && r.ExecutedAt.After(start.Add(-time.Second)) { + applied = append(applied, r.Version) + } + } + + return interfaces.MigrationResult{ + Applied: applied, + DurationMs: time.Since(start).Milliseconds(), + }, nil +} + +// Down rolls back N migrations (Options.Steps, default 1). +// Atlas versioned migrations do not include rollback SQL natively; this driver +// looks for paired _.down.sql files alongside the up migration files. +func (d *Driver) Down(ctx context.Context, req interfaces.MigrationRequest) (interfaces.MigrationResult, error) { + if err := req.Validate(); err != nil { + return interfaces.MigrationResult{}, err + } + start := time.Now() + + db, _, rrw, _, cleanup, err := open(req) + if err != nil { + return interfaces.MigrationResult{}, err + } + defer cleanup() + + steps := req.Options.Steps + if steps <= 0 { + steps = 1 + } + + // Find the last N applied revisions in reverse order. + revs, err := rrw.ReadRevisions(ctx) + if err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas down: read revisions: %w", err) + } + + // Filter to successfully applied revisions. + var applied []*atlmigrate.Revision + for _, r := range revs { + if r.Error == "" { + applied = append(applied, r) + } + } + if len(applied) == 0 { + return interfaces.MigrationResult{DurationMs: time.Since(start).Milliseconds()}, nil + } + + // Roll back the last 'steps' migrations in reverse order. + toRollback := applied + if len(toRollback) > steps { + toRollback = toRollback[len(toRollback)-steps:] + } + + var rolledBack []string + for i := len(toRollback) - 1; i >= 0; i-- { + r := toRollback[i] + if err := runDown(ctx, db, req.Source.Dir, r.Version); err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas down %s: %w", r.Version, err) + } + if err := rrw.DeleteRevision(ctx, r.Version); err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas down: delete revision %s: %w", r.Version, err) + } + rolledBack = append(rolledBack, r.Version) + } + + return interfaces.MigrationResult{ + Applied: rolledBack, + DurationMs: time.Since(start).Milliseconds(), + }, nil +} + +// Status returns the current migration state. +func (d *Driver) Status(ctx context.Context, req interfaces.MigrationRequest) (interfaces.MigrationStatus, error) { + if err := req.Validate(); err != nil { + return interfaces.MigrationStatus{}, err + } + + db, dir, rrw, drv, cleanup, err := open(req) + if err != nil { + return interfaces.MigrationStatus{}, err + } + defer cleanup() + _ = db + _ = drv + + ex, err := atlmigrate.NewExecutor(drv, dir, rrw) + if err != nil { + return interfaces.MigrationStatus{}, fmt.Errorf("atlas status: executor: %w", err) + } + + pending, err := ex.Pending(ctx) + if err != nil && !errors.Is(err, atlmigrate.ErrNoPendingFiles) { + return interfaces.MigrationStatus{}, fmt.Errorf("atlas status: pending: %w", err) + } + + var pendingVersions []string + for _, f := range pending { + pendingVersions = append(pendingVersions, f.Version()) + } + + // Current = latest applied revision with no error. + current, err := rrw.currentVersion(ctx) + if err != nil { + return interfaces.MigrationStatus{}, fmt.Errorf("atlas status: current version: %w", err) + } + + // Check for dirty state (last applied revision has an error). + dirty := false + revs, _ := rrw.ReadRevisions(ctx) + if len(revs) > 0 { + last := revs[len(revs)-1] + dirty = last.Error != "" + } + + return interfaces.MigrationStatus{ + Current: current, + Pending: pendingVersions, + Dirty: dirty, + }, nil +} + +// Goto migrates to the specified version (up or down as needed). +func (d *Driver) Goto(ctx context.Context, req interfaces.MigrationRequest, target string) (interfaces.MigrationResult, error) { + if err := req.Validate(); err != nil { + return interfaces.MigrationResult{}, err + } + start := time.Now() + + db, dir, rrw, drv, cleanup, err := open(req) + if err != nil { + return interfaces.MigrationResult{}, err + } + defer cleanup() + _ = db + + current, err := rrw.currentVersion(ctx) + if err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas goto: current version: %w", err) + } + + // Determine direction. + if target > current { + // Migrate up to target. + ex, err := atlmigrate.NewExecutor(drv, dir, rrw) + if err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas goto: executor: %w", err) + } + if err := ex.ExecuteTo(ctx, target); err != nil && !errors.Is(err, atlmigrate.ErrNoPendingFiles) { + return interfaces.MigrationResult{}, fmt.Errorf("atlas goto %s: %w", target, err) + } + } else if target < current { + // Roll back until we reach target. + revs, err := rrw.ReadRevisions(ctx) + if err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas goto: read revisions: %w", err) + } + for i := len(revs) - 1; i >= 0; i-- { + r := revs[i] + if r.Version <= target { + break + } + if err := runDown(ctx, db, req.Source.Dir, r.Version); err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas goto down %s: %w", r.Version, err) + } + if err := rrw.DeleteRevision(ctx, r.Version); err != nil { + return interfaces.MigrationResult{}, fmt.Errorf("atlas goto: delete revision %s: %w", r.Version, err) + } + } + } + + return interfaces.MigrationResult{ + Applied: []string{target}, + DurationMs: time.Since(start).Milliseconds(), + }, nil +} + +// open builds all the objects needed to talk to the database and the migration dir. +func open(req interfaces.MigrationRequest) ( + db *sql.DB, + dir *atlmigrate.LocalDir, + rrw *sqlRevisionRW, + drv atlmigrate.Driver, + cleanup func(), + err error, +) { + db, err = sql.Open("pgx", req.DSN) + if err != nil { + err = fmt.Errorf("atlas: open db: %w", err) + return + } + + dir, err = openDir(req.Source.Dir) + if err != nil { + _ = db.Close() + err = fmt.Errorf("atlas: open dir: %w", err) + return + } + + rrw, err = newSQLRevisionRW(db, "") + if err != nil { + _ = db.Close() + err = fmt.Errorf("atlas: revisions table: %w", err) + return + } + + drv, err = atlpg.Open(db) + if err != nil { + _ = db.Close() + err = fmt.Errorf("atlas: postgres driver: %w", err) + return + } + + cleanup = func() { _ = db.Close() } + return +} + +// openDir opens a LocalDir, auto-generating atlas.sum if it is absent. +func openDir(path string) (*atlmigrate.LocalDir, error) { + dir, err := atlmigrate.NewLocalDir(path) + if err != nil { + return nil, err + } + // Auto-generate atlas.sum when missing so the driver works without the Atlas CLI. + if err := atlmigrate.Validate(dir); errors.Is(err, atlmigrate.ErrChecksumNotFound) { + files, ferr := dir.Files() + if ferr != nil { + return nil, ferr + } + sum, serr := atlmigrate.NewHashFile(files) + if serr != nil { + return nil, serr + } + if werr := atlmigrate.WriteSumFile(dir, sum); werr != nil { + return nil, werr + } + } else if err != nil { + return nil, err + } + return dir, nil +} + +// runDown executes the .down.sql file paired with a versioned Atlas migration. +// Files are expected to follow the naming: _.down.sql. +func runDown(ctx context.Context, db *sql.DB, dir, version string) error { + // Enumerate .down.sql files matching the version prefix. + localDir, err := atlmigrate.NewLocalDir(dir) + if err != nil { + return err + } + files, err := localDir.Files() + if err != nil { + return err + } + for _, f := range files { + if f.Version() == version { + // Look for a paired .down.sql file. + downName := downFilename(f.Name()) + if downName == "" { + return fmt.Errorf("atlas: no .down.sql file for version %s", version) + } + fh, ferr := localDir.Open(downName) + if ferr != nil { + return fmt.Errorf("atlas: open %s: %w", downName, ferr) + } + defer fh.Close() + buf := make([]byte, 1<<20) // 1 MB max + n, _ := fh.Read(buf) + downSQL := string(buf[:n]) + if downSQL == "" { + return fmt.Errorf("atlas: empty .down.sql for version %s", version) + } + _, err = db.ExecContext(ctx, downSQL) + return err + } + } + return fmt.Errorf("atlas: migration version %s not found in directory", version) +} + +// downFilename returns the .down.sql counterpart of an Atlas migration filename. +// e.g. "20240101_create_users.sql" → "20240101_create_users.down.sql" +// Returns "" if the name doesn't end in .sql or already has .down.sql. +func downFilename(name string) string { + const suffix = ".sql" + const downSuffix = ".down.sql" + if len(name) <= len(suffix) { + return "" + } + if name[len(name)-len(downSuffix):] == downSuffix { + return "" // already a down file + } + if name[len(name)-len(suffix):] != suffix { + return "" + } + return name[:len(name)-len(suffix)] + downSuffix +} diff --git a/internal/atlas/driver_test.go b/internal/atlas/driver_test.go new file mode 100644 index 0000000..25fa0e3 --- /dev/null +++ b/internal/atlas/driver_test.go @@ -0,0 +1,125 @@ +package atlas_test + +import ( + "context" + "os" + "testing" + + atlmigrate "ariga.io/atlas/sql/migrate" + + atlasdriver "github.com/GoCodeAlone/workflow-plugin-migrations/internal/atlas" + "github.com/GoCodeAlone/workflow/interfaces" + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/testharness" +) + +func TestAtlasDriver_Name(t *testing.T) { + d := atlasdriver.New() + if got := d.Name(); got != "atlas" { + t.Errorf("Name() = %q; want %q", got, "atlas") + } +} + +func TestAtlasDriver_UpDownStatus(t *testing.T) { + if testing.Short() { + t.Skip("skipping: requires postgres") + } + h, err := testharness.New() + if err != nil { + t.Skipf("skipping: no postgres available: %v", err) + } + defer h.Close(t) + + dir := makeAtlasDir(t) + ctx := context.Background() + d := atlasdriver.New() + req := interfaces.MigrationRequest{ + DSN: h.DSN(), + Source: interfaces.MigrationSource{Dir: dir}, + } + + // Status: fresh DB — all pending. + st, err := d.Status(ctx, req) + if err != nil { + t.Fatalf("Status() error: %v", err) + } + if len(st.Pending) == 0 { + t.Error("expected pending migrations on fresh DB") + } + + // Up: apply all. + result, err := d.Up(ctx, req) + if err != nil { + t.Fatalf("Up() error: %v", err) + } + if len(result.Applied) != 2 { + t.Errorf("Applied = %v; want 2", result.Applied) + } + + // Status after up: nothing pending. + st, err = d.Status(ctx, req) + if err != nil { + t.Fatalf("Status() after up: %v", err) + } + if len(st.Pending) > 0 { + t.Errorf("unexpected pending after up: %v", st.Pending) + } + if st.Current == "" { + t.Error("expected non-empty Current after up") + } + + // Down: roll back 1. + req.Options.Steps = 1 + _, err = d.Down(ctx, req) + if err != nil { + t.Fatalf("Down() error: %v", err) + } + + st, err = d.Status(ctx, req) + if err != nil { + t.Fatalf("Status() after down: %v", err) + } + if len(st.Pending) == 0 { + t.Error("expected 1 pending migration after rolling back 1") + } +} + +// makeAtlasDir creates a temporary Atlas migration directory with two versioned +// SQL migrations and their down counterparts, plus a valid atlas.sum file. +func makeAtlasDir(t *testing.T) string { + t.Helper() + dir := t.TempDir() + + up1 := []byte("CREATE TABLE users (id SERIAL PRIMARY KEY, name TEXT NOT NULL);\n") + dn1 := []byte("DROP TABLE IF EXISTS users;\n") + up2 := []byte("CREATE TABLE posts (id SERIAL PRIMARY KEY, title TEXT NOT NULL);\n") + dn2 := []byte("DROP TABLE IF EXISTS posts;\n") + + write := func(name string, data []byte) { + t.Helper() + if err := os.WriteFile(dir+"/"+name, data, 0o644); err != nil { + t.Fatal(err) + } + } + write("00001_create_users.sql", up1) + write("00001_create_users.down.sql", dn1) + write("00002_create_posts.sql", up2) + write("00002_create_posts.down.sql", dn2) + + // Generate atlas.sum so the executor validates successfully. + localDir, err := atlmigrate.NewLocalDir(dir) + if err != nil { + t.Fatalf("NewLocalDir: %v", err) + } + files, err := localDir.Files() + if err != nil { + t.Fatalf("dir.Files(): %v", err) + } + sum, err := atlmigrate.NewHashFile(files) + if err != nil { + t.Fatalf("NewHashFile: %v", err) + } + if err := atlmigrate.WriteSumFile(localDir, sum); err != nil { + t.Fatalf("WriteSumFile: %v", err) + } + return dir +} diff --git a/internal/atlas/revisions.go b/internal/atlas/revisions.go new file mode 100644 index 0000000..a713fa3 --- /dev/null +++ b/internal/atlas/revisions.go @@ -0,0 +1,164 @@ +package atlas + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" + + atlmigrate "ariga.io/atlas/sql/migrate" +) + +const defaultRevisionsTable = "atlas_schema_revisions" + +// sqlRevisionRW is a database-backed RevisionReadWriter for Atlas migrations. +// It creates and manages an atlas_schema_revisions table in the target schema. +type sqlRevisionRW struct { + db *sql.DB + table string +} + +// newSQLRevisionRW returns a sqlRevisionRW backed by the given *sql.DB. +func newSQLRevisionRW(db *sql.DB, table string) (*sqlRevisionRW, error) { + if table == "" { + table = defaultRevisionsTable + } + rw := &sqlRevisionRW{db: db, table: table} + if err := rw.init(context.Background()); err != nil { + return nil, err + } + return rw, nil +} + +// Ident satisfies RevisionReadWriter. +func (r *sqlRevisionRW) Ident() *atlmigrate.TableIdent { + return &atlmigrate.TableIdent{Name: r.table} +} + +// init creates the revisions table if it does not exist. +func (r *sqlRevisionRW) init(ctx context.Context) error { + _, err := r.db.ExecContext(ctx, fmt.Sprintf(` + CREATE TABLE IF NOT EXISTS %q ( + version TEXT NOT NULL PRIMARY KEY, + description TEXT NOT NULL DEFAULT '', + type INTEGER NOT NULL DEFAULT 0, + applied INTEGER NOT NULL DEFAULT 0, + total INTEGER NOT NULL DEFAULT 0, + executed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + execution_time BIGINT NOT NULL DEFAULT 0, + error TEXT, + error_stmt TEXT, + hash TEXT NOT NULL DEFAULT '', + operator_version TEXT NOT NULL DEFAULT '' + )`, r.table)) + return err +} + +// ReadRevisions returns all applied revisions ordered by version. +func (r *sqlRevisionRW) ReadRevisions(ctx context.Context) ([]*atlmigrate.Revision, error) { + rows, err := r.db.QueryContext(ctx, fmt.Sprintf( + `SELECT version, description, type, applied, total, executed_at, execution_time, + COALESCE(error,''), COALESCE(error_stmt,''), hash, operator_version + FROM %q ORDER BY version`, r.table)) + if err != nil { + return nil, err + } + defer rows.Close() + return scanRevisions(rows) +} + +// ReadRevision returns the revision for a specific version. +func (r *sqlRevisionRW) ReadRevision(ctx context.Context, v string) (*atlmigrate.Revision, error) { + row := r.db.QueryRowContext(ctx, fmt.Sprintf( + `SELECT version, description, type, applied, total, executed_at, execution_time, + COALESCE(error,''), COALESCE(error_stmt,''), hash, operator_version + FROM %q WHERE version = $1`, r.table), v) + rev, err := scanRevision(row) + if errors.Is(err, sql.ErrNoRows) { + return nil, atlmigrate.ErrRevisionNotExist + } + return rev, err +} + +// WriteRevision upserts a revision record. +func (r *sqlRevisionRW) WriteRevision(ctx context.Context, rev *atlmigrate.Revision) error { + _, err := r.db.ExecContext(ctx, fmt.Sprintf(` + INSERT INTO %q (version, description, type, applied, total, executed_at, execution_time, + error, error_stmt, hash, operator_version) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) + ON CONFLICT (version) DO UPDATE SET + description = EXCLUDED.description, + type = EXCLUDED.type, + applied = EXCLUDED.applied, + total = EXCLUDED.total, + executed_at = EXCLUDED.executed_at, + execution_time = EXCLUDED.execution_time, + error = EXCLUDED.error, + error_stmt = EXCLUDED.error_stmt, + hash = EXCLUDED.hash, + operator_version = EXCLUDED.operator_version`, r.table), + rev.Version, rev.Description, int(rev.Type), rev.Applied, rev.Total, + rev.ExecutedAt, int64(rev.ExecutionTime), + rev.Error, rev.ErrorStmt, rev.Hash, rev.OperatorVersion, + ) + return err +} + +// DeleteRevision removes a revision record by version. +func (r *sqlRevisionRW) DeleteRevision(ctx context.Context, v string) error { + _, err := r.db.ExecContext(ctx, + fmt.Sprintf(`DELETE FROM %q WHERE version = $1`, r.table), v) + return err +} + +// currentVersion returns the latest applied version, or "" if none. +func (r *sqlRevisionRW) currentVersion(ctx context.Context) (string, error) { + var v string + err := r.db.QueryRowContext(ctx, + fmt.Sprintf(`SELECT version FROM %q WHERE error = '' ORDER BY version DESC LIMIT 1`, r.table)). + Scan(&v) + if errors.Is(err, sql.ErrNoRows) { + return "", nil + } + return v, err +} + +func scanRevisions(rows *sql.Rows) ([]*atlmigrate.Revision, error) { + var revs []*atlmigrate.Revision + for rows.Next() { + rev := &atlmigrate.Revision{} + var execNs int64 + var executedAt time.Time + var revType int + if err := rows.Scan( + &rev.Version, &rev.Description, &revType, &rev.Applied, &rev.Total, + &executedAt, &execNs, &rev.Error, &rev.ErrorStmt, &rev.Hash, &rev.OperatorVersion, + ); err != nil { + return nil, err + } + rev.Type = atlmigrate.RevisionType(revType) + rev.ExecutedAt = executedAt + rev.ExecutionTime = time.Duration(execNs) + revs = append(revs, rev) + } + return revs, rows.Err() +} + +func scanRevision(row *sql.Row) (*atlmigrate.Revision, error) { + rev := &atlmigrate.Revision{} + var execNs int64 + var executedAt time.Time + var revType int + err := row.Scan( + &rev.Version, &rev.Description, &revType, &rev.Applied, &rev.Total, + &executedAt, &execNs, &rev.Error, &rev.ErrorStmt, &rev.Hash, &rev.OperatorVersion, + ) + if err != nil { + return nil, err + } + rev.Type = atlmigrate.RevisionType(revType) + rev.ExecutedAt = executedAt + rev.ExecutionTime = time.Duration(execNs) + return rev, nil +} diff --git a/internal/atlasplugin/modules.go b/internal/atlasplugin/modules.go new file mode 100644 index 0000000..df2ff5a --- /dev/null +++ b/internal/atlasplugin/modules.go @@ -0,0 +1,149 @@ +package atlasplugin + +import ( + "context" + "fmt" + "os" + "time" + + sdk "github.com/GoCodeAlone/workflow/plugin/external/sdk" + + atlasdriver "github.com/GoCodeAlone/workflow-plugin-migrations/internal/atlas" + "github.com/GoCodeAlone/workflow/interfaces" +) + +// atlasMigrationsModule is a sdk.ModuleInstance for database.migrations backed by Atlas. +type atlasMigrationsModule struct { + name string + sourceDir string + dsnEnv string + timeout time.Duration +} + +func newAtlasMigrationsModule(name string, rawCfg map[string]any) sdk.ModuleInstance { + m := &atlasMigrationsModule{ + name: name, + dsnEnv: "DATABASE_URL", + timeout: 5 * time.Minute, + } + if v, ok := rawCfg["source_dir"].(string); ok { + m.sourceDir = v + } + if v, ok := rawCfg["dsn_env"].(string); ok && v != "" { + m.dsnEnv = v + } + if v, ok := rawCfg["timeout"].(string); ok { + if d, err := time.ParseDuration(v); err == nil { + m.timeout = d + } + } + return m +} + +func (m *atlasMigrationsModule) Init() error { + if m.sourceDir == "" { + return fmt.Errorf("database.migrations %q: source_dir is required", m.name) + } + return nil +} + +func (m *atlasMigrationsModule) Start(_ context.Context) error { return nil } +func (m *atlasMigrationsModule) Stop(_ context.Context) error { return nil } + +func (m *atlasMigrationsModule) InvokeMethod(method string, args map[string]any) (map[string]any, error) { + dsn := "" + if v, ok := args["dsn"].(string); ok && v != "" { + dsn = v + } + if dsn == "" && m.dsnEnv != "" { + dsn = os.Getenv(m.dsnEnv) + } + if dsn == "" { + return nil, fmt.Errorf("database.migrations %q: no DSN available", m.name) + } + + sourceDir := m.sourceDir + if v, ok := args["source_dir"].(string); ok && v != "" { + sourceDir = v + } + + ctx, cancel := context.WithTimeout(context.Background(), m.timeout) + defer cancel() + + d := atlasdriver.New() + req := interfaces.MigrationRequest{ + DSN: dsn, + Source: interfaces.MigrationSource{Dir: sourceDir}, + } + + switch method { + case "up": + result, err := d.Up(ctx, req) + if err != nil { + return nil, err + } + return map[string]any{"applied": result.Applied, "duration_ms": result.DurationMs}, nil + case "down": + steps := 1 + if v, ok := args["steps"].(int); ok { + steps = v + } + req.Options.Steps = steps + result, err := d.Down(ctx, req) + if err != nil { + return nil, err + } + return map[string]any{"applied": result.Applied, "duration_ms": result.DurationMs}, nil + case "status": + st, err := d.Status(ctx, req) + if err != nil { + return nil, err + } + return map[string]any{"current": st.Current, "pending": st.Pending, "dirty": st.Dirty}, nil + case "goto": + target, _ := args["target"].(string) + if target == "" { + return nil, fmt.Errorf("goto: target version is required") + } + result, err := d.Goto(ctx, req, target) + if err != nil { + return nil, err + } + return map[string]any{"applied": result.Applied, "duration_ms": result.DurationMs}, nil + default: + return nil, fmt.Errorf("database.migrations %q: unknown method %q", m.name, method) + } +} + +// driverModule is a sdk.ModuleInstance for database.migration_driver (atlas). +type driverModule struct { + driver interfaces.MigrationDriver +} + +func (m *driverModule) Init() error { return nil } +func (m *driverModule) Start(_ context.Context) error { return nil } +func (m *driverModule) Stop(_ context.Context) error { return nil } +func (m *driverModule) InvokeMethod(method string, _ map[string]any) (map[string]any, error) { + if method == "driver_name" { + return map[string]any{"name": m.driver.Name()}, nil + } + return nil, fmt.Errorf("database.migration_driver: unknown method %q", method) +} + +// driverBackedModule wraps a non-Atlas driver (golang-migrate, goose) as a module. +type driverBackedModule struct { + name string + rawCfg map[string]any + driver interfaces.MigrationDriver +} + +func newDriverBackedModule(name string, rawCfg map[string]any, d interfaces.MigrationDriver) sdk.ModuleInstance { + return &driverBackedModule{name: name, rawCfg: rawCfg, driver: d} +} + +func (m *driverBackedModule) Init() error { return nil } +func (m *driverBackedModule) Start(_ context.Context) error { return nil } +func (m *driverBackedModule) Stop(_ context.Context) error { return nil } +func (m *driverBackedModule) InvokeMethod(method string, args map[string]any) (map[string]any, error) { + return nil, fmt.Errorf("driverBackedModule %q: method %q not yet implemented in atlas plugin", m.name, method) +} diff --git a/internal/atlasplugin/plugin.go b/internal/atlasplugin/plugin.go index 9d60693..7daacd7 100644 --- a/internal/atlasplugin/plugin.go +++ b/internal/atlasplugin/plugin.go @@ -1,12 +1,16 @@ // Package atlasplugin provides the Atlas migration driver plugin. -// Atlas has larger dependencies (HCL toolchain) so it ships as a separate binary. -// This implementation will be fleshed out in Task 27. +// Atlas has larger dependencies (HCL toolchain) so it ships as a separate binary +// to keep the main workflow-plugin-migrations binary lean. package atlasplugin import ( "fmt" sdk "github.com/GoCodeAlone/workflow/plugin/external/sdk" + + atlasdriver "github.com/GoCodeAlone/workflow-plugin-migrations/internal/atlas" + "github.com/GoCodeAlone/workflow-plugin-migrations/internal/golangmigrate" + "github.com/GoCodeAlone/workflow-plugin-migrations/internal/goose" ) // Version is set at build time via -ldflags. @@ -26,24 +30,49 @@ func (p *AtlasPlugin) Manifest() sdk.PluginManifest { Name: "workflow-plugin-atlas-migrate", Version: Version, Author: "GoCodeAlone", - Description: "Atlas HCL migration driver for the workflow engine", + Description: "Atlas migration driver for the workflow engine (also ships golang-migrate + goose)", } } // ModuleTypes returns the module types this plugin provides. func (p *AtlasPlugin) ModuleTypes() []string { - return []string{"database.migrations"} + return []string{"database.migrations", "database.migration_driver"} } -// CreateModule creates a module instance (atlas driver only). -func (p *AtlasPlugin) CreateModule(typeName, name string, _ map[string]any) (sdk.ModuleInstance, error) { - return nil, fmt.Errorf("atlasplugin: driver implementation coming in Task 27 (use workflow-plugin-migrations binary for golang-migrate/goose)") +// CreateModule creates a module instance for the requested type. +// This binary ships all three drivers (golang-migrate, goose, atlas) but the +// atlas driver is the primary addition over the main plugin binary. +func (p *AtlasPlugin) CreateModule(typeName, name string, rawCfg map[string]any) (sdk.ModuleInstance, error) { + switch typeName { + case "database.migrations": + driverName, _ := rawCfg["driver"].(string) + if driverName == "" { + driverName, _ = rawCfg["driver_ref"].(string) + } + if driverName == "" { + driverName = "atlas" + } + switch driverName { + case "atlas": + return newAtlasMigrationsModule(name, rawCfg), nil + case "golang-migrate": + return newDriverBackedModule(name, rawCfg, golangmigrate.New()), nil + case "goose": + return newDriverBackedModule(name, rawCfg, goose.New()), nil + default: + return nil, fmt.Errorf("atlasplugin: unknown driver %q (supported: atlas, golang-migrate, goose)", driverName) + } + case "database.migration_driver": + return &driverModule{driver: atlasdriver.New()}, nil + default: + return nil, fmt.Errorf("atlasplugin: unknown module type %q", typeName) + } } -// StepTypes returns step type names. +// StepTypes returns nil — steps are handled by the main plugin binary. func (p *AtlasPlugin) StepTypes() []string { return nil } -// CreateStep creates a step instance. +// CreateStep returns an error — use the main plugin binary for pipeline steps. func (p *AtlasPlugin) CreateStep(typeName, name string, _ map[string]any) (sdk.StepInstance, error) { - return nil, fmt.Errorf("atlasplugin: no steps provided") + return nil, fmt.Errorf("atlasplugin: pipeline steps are provided by workflow-plugin-migrations, not workflow-plugin-atlas-migrate") } diff --git a/pkg/cli/lint.go b/pkg/cli/lint.go new file mode 100644 index 0000000..464dc38 --- /dev/null +++ b/pkg/cli/lint.go @@ -0,0 +1,67 @@ +package cli + +import ( + "encoding/json" + "fmt" + "os" + + "github.com/spf13/cobra" + + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/lint" +) + +func newLintCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "lint ", + Short: "Static analysis for migration files", + Long: `Lint checks a migration directory for common issues: + L001 - version ordering gap + L002 - duplicate version + L003 - naming convention violation + L004 - dangerous SQL operation without safety guard + L005 - missing .down.sql counterpart (paired format) + L006 - empty migration file`, + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + dir := args[0] + jsonOut, _ := cmd.Flags().GetBool("json") + + result, err := lint.Lint(dir) + if err != nil { + return fmt.Errorf("lint: %w", err) + } + + if jsonOut { + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + return enc.Encode(result) + } + + if result.Clean { + fmt.Println("✓ No issues found.") + return nil + } + + for _, issue := range result.Issues { + loc := issue.File + if loc == "" && issue.Version != "" { + loc = "v" + issue.Version + } + fmt.Printf("[%s] %s (%s): %s\n", issue.Severity, issue.Code, loc, issue.Message) + } + + errCount := 0 + for _, i := range result.Issues { + if i.Severity == lint.SeverityError { + errCount++ + } + } + if errCount > 0 { + return fmt.Errorf("lint: %d error(s) found", errCount) + } + return nil + }, + } + cmd.Flags().Bool("json", false, "Output results as JSON") + return cmd +} diff --git a/pkg/cli/root.go b/pkg/cli/root.go index 79d1004..862f3f5 100644 --- a/pkg/cli/root.go +++ b/pkg/cli/root.go @@ -12,6 +12,7 @@ import ( "github.com/GoCodeAlone/workflow/interfaces" + atlasdriver "github.com/GoCodeAlone/workflow-plugin-migrations/internal/atlas" "github.com/GoCodeAlone/workflow-plugin-migrations/internal/golangmigrate" "github.com/GoCodeAlone/workflow-plugin-migrations/internal/goose" ) @@ -38,20 +39,21 @@ func NewRoot() *cobra.Command { root := &cobra.Command{ Use: "migrate", Short: "Database migration commands", - Long: "Run, inspect, and test database migrations via golang-migrate or goose.", + Long: "Run, inspect, and test database migrations via golang-migrate, goose, or atlas.", } root.AddCommand( newUpCmd(), newDownCmd(), newStatusCmd(), newGotoCmd(), + newLintCmd(), ) return root } // sharedFlags adds the common driver/DSN/source-dir flags to a command. func sharedFlags(cmd *cobra.Command) { - cmd.Flags().String("driver", "golang-migrate", "Migration driver (golang-migrate|goose)") + cmd.Flags().String("driver", "golang-migrate", "Migration driver (golang-migrate|goose|atlas)") cmd.Flags().String("source-dir", "", "Directory containing migration files (required)") cmd.Flags().String("dsn", "", "Database connection string (overrides DATABASE_URL env var)") _ = cmd.MarkFlagRequired("source-dir") @@ -75,8 +77,10 @@ func buildDriverAndRequest(cmd *cobra.Command) (interfaces.MigrationDriver, inte d = golangmigrate.New() case "goose": d = goose.New() + case "atlas": + d = atlasdriver.New() default: - return nil, interfaces.MigrationRequest{}, fmt.Errorf("unknown driver %q (supported: golang-migrate, goose)", driverName) + return nil, interfaces.MigrationRequest{}, fmt.Errorf("unknown driver %q (supported: golang-migrate, goose, atlas)", driverName) } req := interfaces.MigrationRequest{ diff --git a/pkg/lint/lint.go b/pkg/lint/lint.go new file mode 100644 index 0000000..1274030 --- /dev/null +++ b/pkg/lint/lint.go @@ -0,0 +1,228 @@ +// Package lint provides static analysis for database migration directories. +// It checks for ordering gaps, duplicate versions, naming convention violations, +// dangerous SQL operations, and missing down migrations. +package lint + +import ( + "fmt" + "os" + "path/filepath" + "regexp" + "sort" + "strconv" + "strings" +) + +// Severity levels for lint issues. +const ( + SeverityError = "error" + SeverityWarning = "warning" +) + +// Issue describes a single lint finding. +type Issue struct { + File string `json:"file"` + Version string `json:"version,omitempty"` + Severity string `json:"severity"` + Code string `json:"code"` + Message string `json:"message"` +} + +// Result is the outcome of linting a migration directory. +type Result struct { + Dir string `json:"dir"` + Issues []Issue `json:"issues"` + Clean bool `json:"clean"` +} + +// migrationFile represents a parsed migration file entry. +type migrationFile struct { + name string + version uint64 + isUp bool + isDown bool + isAtlas bool // no .up/.down suffix (atlas or goose format) +} + +var ( + // pairedRe matches: _.up.sql or _.down.sql + pairedRe = regexp.MustCompile(`^(\d+)_(.+)\.(up|down)\.sql$`) + // atlasRe matches: _.sql (atlas/goose combined format) + atlasRe = regexp.MustCompile(`^(\d+)_(.+)\.sql$`) + + // dangerousOps are SQL patterns that may cause data loss. + dangerousPatterns = []*regexp.Regexp{ + regexp.MustCompile(`(?i)\bDROP\s+TABLE\b`), + regexp.MustCompile(`(?i)\bDROP\s+COLUMN\b`), + regexp.MustCompile(`(?i)\bTRUNCATE\b`), + regexp.MustCompile(`(?i)\bDROP\s+SCHEMA\b`), + regexp.MustCompile(`(?i)\bDELETE\s+FROM\b`), + } + // safeGuards indicate the dangerous op is intentional and guarded. + safeGuards = []*regexp.Regexp{ + regexp.MustCompile(`(?i)\bIF\s+EXISTS\b`), + regexp.MustCompile(`(?i)\bWHERE\b`), // DELETE FROM ... WHERE + } +) + +// Lint analyses the migration directory at dir and returns a Result. +func Lint(dir string) (*Result, error) { + entries, err := os.ReadDir(dir) + if err != nil { + return nil, fmt.Errorf("lint: read dir %s: %w", dir, err) + } + + result := &Result{Dir: dir} + + // Parse all .sql files. + byVersion := map[uint64]*migrationFile{} + var versions []uint64 + + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") { + continue + } + name := e.Name() + + var mf *migrationFile + + if m := pairedRe.FindStringSubmatch(name); m != nil { + v, _ := strconv.ParseUint(m[1], 10, 64) + existing, ok := byVersion[v] + if !ok { + existing = &migrationFile{version: v} + byVersion[v] = existing + versions = append(versions, v) + } + mf = existing + mf.name = name + if m[3] == "up" { + mf.isUp = true + } else { + mf.isDown = true + } + } else if m := atlasRe.FindStringSubmatch(name); m != nil { + v, _ := strconv.ParseUint(m[1], 10, 64) + if _, ok := byVersion[v]; !ok { + byVersion[v] = &migrationFile{version: v, name: name, isAtlas: true} + versions = append(versions, v) + } + mf = byVersion[v] + } else { + result.Issues = append(result.Issues, Issue{ + File: name, + Severity: SeverityWarning, + Code: "L003", + Message: fmt.Sprintf("file %q does not match expected naming convention (_.up.sql or _.sql)", name), + }) + continue + } + _ = mf + + // L004: dangerous operations. + content, rerr := os.ReadFile(filepath.Join(dir, name)) + if rerr != nil { + continue + } + result.Issues = append(result.Issues, checkDangerous(name, string(content))...) + } + + // Sort versions for ordered checks. + sort.Slice(versions, func(i, j int) bool { return versions[i] < versions[j] }) + + // L001: ordering gaps. + for i := 1; i < len(versions); i++ { + if versions[i] != versions[i-1]+1 { + result.Issues = append(result.Issues, Issue{ + Version: fmt.Sprintf("%d", versions[i]), + Severity: SeverityWarning, + Code: "L001", + Message: fmt.Sprintf("version gap: no migration between %d and %d", versions[i-1], versions[i]), + }) + } + } + + // L002: duplicate versions — already handled (map dedup), but check filenames for exact dupes. + seen := map[string]bool{} + for _, e := range entries { + if seen[e.Name()] { + result.Issues = append(result.Issues, Issue{ + File: e.Name(), + Severity: SeverityError, + Code: "L002", + Message: fmt.Sprintf("duplicate migration file: %s", e.Name()), + }) + } + seen[e.Name()] = true + } + + // L005: missing down files for paired-format migrations. + for _, v := range versions { + mf := byVersion[v] + if mf.isAtlas { + continue // atlas/goose format — single file, down is embedded or separate tool + } + if mf.isUp && !mf.isDown { + result.Issues = append(result.Issues, Issue{ + Version: fmt.Sprintf("%d", v), + Severity: SeverityWarning, + Code: "L005", + Message: fmt.Sprintf("migration v%d has no corresponding .down.sql file", v), + }) + } + } + + // L006: empty files. + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".sql") { + continue + } + info, _ := e.Info() + if info != nil && info.Size() == 0 { + result.Issues = append(result.Issues, Issue{ + File: e.Name(), + Severity: SeverityError, + Code: "L006", + Message: fmt.Sprintf("migration file %q is empty", e.Name()), + }) + } + } + + result.Clean = len(result.Issues) == 0 + return result, nil +} + +// checkDangerous scans SQL content for potentially dangerous operations that +// lack safety guards (IF EXISTS, WHERE clauses, etc.). +func checkDangerous(filename, content string) []Issue { + var issues []Issue + lines := strings.Split(content, "\n") + for i, line := range lines { + trimmed := strings.TrimSpace(line) + if strings.HasPrefix(trimmed, "--") { + continue // skip comments + } + for _, pat := range dangerousPatterns { + if !pat.MatchString(line) { + continue + } + // Check if any safe guard is present on the same line. + guarded := false + for _, guard := range safeGuards { + if guard.MatchString(line) { + guarded = true + break + } + } + if !guarded { + issues = append(issues, Issue{ + File: filename, + Severity: SeverityWarning, + Code: "L004", + Message: fmt.Sprintf("line %d: potentially dangerous operation without safety guard: %s", i+1, strings.TrimSpace(line)), + }) + } + } + } + return issues +} diff --git a/pkg/lint/lint_test.go b/pkg/lint/lint_test.go new file mode 100644 index 0000000..5880053 --- /dev/null +++ b/pkg/lint/lint_test.go @@ -0,0 +1,121 @@ +package lint_test + +import ( + "os" + "path/filepath" + "testing" + + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/lint" +) + +func TestLint_Clean(t *testing.T) { + dir := t.TempDir() + write(t, dir, "001_create_users.up.sql", "CREATE TABLE users (id SERIAL PRIMARY KEY);") + write(t, dir, "001_create_users.down.sql", "DROP TABLE IF EXISTS users;") + write(t, dir, "002_add_email.up.sql", "ALTER TABLE users ADD COLUMN email TEXT;") + write(t, dir, "002_add_email.down.sql", "ALTER TABLE users DROP COLUMN IF EXISTS email;") + + r, err := lint.Lint(dir) + if err != nil { + t.Fatalf("Lint error: %v", err) + } + if !r.Clean { + t.Errorf("expected clean result, got issues: %+v", r.Issues) + } +} + +func TestLint_OrderingGap(t *testing.T) { + dir := t.TempDir() + write(t, dir, "001_create_users.up.sql", "CREATE TABLE users (id SERIAL PRIMARY KEY);") + write(t, dir, "001_create_users.down.sql", "DROP TABLE IF EXISTS users;") + // skip 002 + write(t, dir, "003_add_email.up.sql", "ALTER TABLE users ADD COLUMN email TEXT;") + write(t, dir, "003_add_email.down.sql", "ALTER TABLE users DROP COLUMN IF EXISTS email;") + + r, err := lint.Lint(dir) + if err != nil { + t.Fatalf("Lint error: %v", err) + } + assertIssue(t, r, "L001") +} + +func TestLint_MissingDown(t *testing.T) { + dir := t.TempDir() + write(t, dir, "001_create_users.up.sql", "CREATE TABLE users (id SERIAL PRIMARY KEY);") + // no .down.sql + + r, err := lint.Lint(dir) + if err != nil { + t.Fatalf("Lint error: %v", err) + } + assertIssue(t, r, "L005") +} + +func TestLint_NamingConvention(t *testing.T) { + dir := t.TempDir() + write(t, dir, "CreateUsers.sql", "CREATE TABLE users (id SERIAL PRIMARY KEY);") + + r, err := lint.Lint(dir) + if err != nil { + t.Fatalf("Lint error: %v", err) + } + assertIssue(t, r, "L003") +} + +func TestLint_DangerousOp(t *testing.T) { + dir := t.TempDir() + write(t, dir, "001_drop_users.up.sql", "DROP TABLE users;") // no IF EXISTS + write(t, dir, "001_drop_users.down.sql", "CREATE TABLE users (id SERIAL PRIMARY KEY);") + + r, err := lint.Lint(dir) + if err != nil { + t.Fatalf("Lint error: %v", err) + } + assertIssue(t, r, "L004") +} + +func TestLint_DangerousOpSafe(t *testing.T) { + dir := t.TempDir() + write(t, dir, "001_drop_users.up.sql", "DROP TABLE IF EXISTS users;") + write(t, dir, "001_drop_users.down.sql", "CREATE TABLE users (id SERIAL PRIMARY KEY);") + + r, err := lint.Lint(dir) + if err != nil { + t.Fatalf("Lint error: %v", err) + } + // L004 should NOT fire since IF EXISTS is present. + for _, issue := range r.Issues { + if issue.Code == "L004" { + t.Errorf("unexpected L004 issue for guarded DROP TABLE IF EXISTS: %+v", issue) + } + } +} + +func TestLint_EmptyFile(t *testing.T) { + dir := t.TempDir() + write(t, dir, "001_create_users.up.sql", "") + write(t, dir, "001_create_users.down.sql", "DROP TABLE IF EXISTS users;") + + r, err := lint.Lint(dir) + if err != nil { + t.Fatalf("Lint error: %v", err) + } + assertIssue(t, r, "L006") +} + +func assertIssue(t *testing.T, r *lint.Result, code string) { + t.Helper() + for _, issue := range r.Issues { + if issue.Code == code { + return + } + } + t.Errorf("expected issue %s not found in: %+v", code, r.Issues) +} + +func write(t *testing.T, dir, name, content string) { + t.Helper() + if err := os.WriteFile(filepath.Join(dir, name), []byte(content), 0o644); err != nil { + t.Fatal(err) + } +} diff --git a/plugin.json b/plugin.json index e0e5a53..bddb8c2 100644 --- a/plugin.json +++ b/plugin.json @@ -1,14 +1,14 @@ { "name": "workflow-plugin-migrations", "version": "0.0.0", - "description": "Database migration plugin for the workflow engine: golang-migrate + goose drivers, pre-deploy runner, wfctl migrate CLI", + "description": "Database migration plugin for the workflow engine: golang-migrate + goose + atlas drivers, pre-deploy runner, wfctl migrate CLI, static lint tool", "author": "GoCodeAlone", "license": "Apache-2.0", "type": "external", "tier": "community", "private": false, "minEngineVersion": "0.18.0", - "keywords": ["migrations", "database", "postgres", "golang-migrate", "goose"], + "keywords": ["migrations", "database", "postgres", "golang-migrate", "goose", "atlas"], "homepage": "https://github.com/GoCodeAlone/workflow-plugin-migrations", "repository": "https://github.com/GoCodeAlone/workflow-plugin-migrations", "capabilities": { @@ -28,12 +28,13 @@ "cliCommands": [ { "name": "migrate", - "description": "Run database migrations", + "description": "Run and inspect database migrations", "subcommands": [ {"name": "up", "description": "Apply all pending migrations"}, {"name": "down", "description": "Roll back migrations"}, {"name": "status", "description": "Show migration status"}, - {"name": "goto", "description": "Migrate to a specific version"} + {"name": "goto", "description": "Migrate to a specific version"}, + {"name": "lint", "description": "Static analysis for migration files"} ], "flags_passthrough": true } From 23bcad7fb8d631f6570be0583a8186a275e6d22e Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 22 Apr 2026 21:44:36 -0400 Subject: [PATCH 2/6] style+fix: address code-reviewer feedback on 399a7ea - gofmt: fix import ordering in driver_test.go (workflow-plugin-migrations before workflow) - gofmt: fix over-aligned stubs in atlasplugin/modules.go - atlas/driver.go: replace 1MB fixed read buffer with io.ReadAll in runDown - Dockerfile: use ARG TARGETARCH + GOARCH=$TARGETARCH for multi-platform buildx support - atlasplugin/modules.go: add clarifying comment explaining driverBackedModule.InvokeMethod is intentionally unimplemented (use workflow-plugin-migrations binary for those drivers) Co-Authored-By: Claude Sonnet 4.6 --- cmd/workflow-migrate/Dockerfile | 5 +++-- internal/atlas/driver.go | 9 ++++++--- internal/atlas/driver_test.go | 2 +- internal/atlasplugin/modules.go | 24 ++++++++++++++++-------- 4 files changed, 26 insertions(+), 14 deletions(-) diff --git a/cmd/workflow-migrate/Dockerfile b/cmd/workflow-migrate/Dockerfile index 3b022a2..df0c7ab 100644 --- a/cmd/workflow-migrate/Dockerfile +++ b/cmd/workflow-migrate/Dockerfile @@ -1,11 +1,12 @@ # syntax=docker/dockerfile:1 # Build stage -FROM golang:1.26-alpine AS builder +FROM --platform=$BUILDPLATFORM golang:1.26-alpine AS builder +ARG TARGETARCH WORKDIR /src COPY go.mod go.sum ./ RUN go mod download COPY . . -RUN CGO_ENABLED=0 GOARCH=amd64 GOOS=linux \ +RUN CGO_ENABLED=0 GOARCH=$TARGETARCH GOOS=linux \ go build -ldflags="-s -w" -o /workflow-migrate ./cmd/workflow-migrate # Runtime stage — distroless/static has no shell, minimal attack surface diff --git a/internal/atlas/driver.go b/internal/atlas/driver.go index 712ae2c..a5cc2f7 100644 --- a/internal/atlas/driver.go +++ b/internal/atlas/driver.go @@ -9,6 +9,7 @@ import ( "database/sql" "errors" "fmt" + "io" "time" atlmigrate "ariga.io/atlas/sql/migrate" @@ -326,9 +327,11 @@ func runDown(ctx context.Context, db *sql.DB, dir, version string) error { return fmt.Errorf("atlas: open %s: %w", downName, ferr) } defer fh.Close() - buf := make([]byte, 1<<20) // 1 MB max - n, _ := fh.Read(buf) - downSQL := string(buf[:n]) + buf, rerr := io.ReadAll(fh) + if rerr != nil { + return fmt.Errorf("atlas: read %s: %w", downName, rerr) + } + downSQL := string(buf) if downSQL == "" { return fmt.Errorf("atlas: empty .down.sql for version %s", version) } diff --git a/internal/atlas/driver_test.go b/internal/atlas/driver_test.go index 25fa0e3..1f2b4ac 100644 --- a/internal/atlas/driver_test.go +++ b/internal/atlas/driver_test.go @@ -8,8 +8,8 @@ import ( atlmigrate "ariga.io/atlas/sql/migrate" atlasdriver "github.com/GoCodeAlone/workflow-plugin-migrations/internal/atlas" - "github.com/GoCodeAlone/workflow/interfaces" "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/testharness" + "github.com/GoCodeAlone/workflow/interfaces" ) func TestAtlasDriver_Name(t *testing.T) { diff --git a/internal/atlasplugin/modules.go b/internal/atlasplugin/modules.go index df2ff5a..9f3e604 100644 --- a/internal/atlasplugin/modules.go +++ b/internal/atlasplugin/modules.go @@ -120,9 +120,9 @@ type driverModule struct { driver interfaces.MigrationDriver } -func (m *driverModule) Init() error { return nil } -func (m *driverModule) Start(_ context.Context) error { return nil } -func (m *driverModule) Stop(_ context.Context) error { return nil } +func (m *driverModule) Init() error { return nil } +func (m *driverModule) Start(_ context.Context) error { return nil } +func (m *driverModule) Stop(_ context.Context) error { return nil } func (m *driverModule) InvokeMethod(method string, _ map[string]any) (map[string]any, error) { if method == "driver_name" { return map[string]any{"name": m.driver.Name()}, nil @@ -141,9 +141,17 @@ func newDriverBackedModule(name string, rawCfg map[string]any, d interfaces.Migr return &driverBackedModule{name: name, rawCfg: rawCfg, driver: d} } -func (m *driverBackedModule) Init() error { return nil } -func (m *driverBackedModule) Start(_ context.Context) error { return nil } -func (m *driverBackedModule) Stop(_ context.Context) error { return nil } -func (m *driverBackedModule) InvokeMethod(method string, args map[string]any) (map[string]any, error) { - return nil, fmt.Errorf("driverBackedModule %q: method %q not yet implemented in atlas plugin", m.name, method) +func (m *driverBackedModule) Init() error { return nil } +func (m *driverBackedModule) Start(_ context.Context) error { return nil } +func (m *driverBackedModule) Stop(_ context.Context) error { return nil } +// InvokeMethod is intentionally unimplemented for golang-migrate and goose modules +// created via the atlas binary. Use the main workflow-plugin-migrations binary +// (which ships without Atlas HCL dependencies) for those drivers. This module +// exists only so that CreateModule can return a non-nil instance for config +// validation; actual execution routes through the appropriate binary. +func (m *driverBackedModule) InvokeMethod(method string, _ map[string]any) (map[string]any, error) { + return nil, fmt.Errorf( + "database.migrations %q: driver %q methods must be invoked via workflow-plugin-migrations, not workflow-plugin-atlas-migrate", + m.name, m.driver.Name(), + ) } From 24bb1729456f976d8c6092c0b0fc01935b99fd82 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 22 Apr 2026 21:51:16 -0400 Subject: [PATCH 3/6] style: add blank line before InvokeMethod doc comment in modules.go gofmt requires a blank line between the Stop stub and the doc comment on driverBackedModule.InvokeMethod. Co-Authored-By: Claude Sonnet 4.6 --- internal/atlasplugin/modules.go | 1 + 1 file changed, 1 insertion(+) diff --git a/internal/atlasplugin/modules.go b/internal/atlasplugin/modules.go index 9f3e604..574c1db 100644 --- a/internal/atlasplugin/modules.go +++ b/internal/atlasplugin/modules.go @@ -144,6 +144,7 @@ func newDriverBackedModule(name string, rawCfg map[string]any, d interfaces.Migr func (m *driverBackedModule) Init() error { return nil } func (m *driverBackedModule) Start(_ context.Context) error { return nil } func (m *driverBackedModule) Stop(_ context.Context) error { return nil } + // InvokeMethod is intentionally unimplemented for golang-migrate and goose modules // created via the atlas binary. Use the main workflow-plugin-migrations binary // (which ships without Atlas HCL dependencies) for those drivers. This module From dd51908e13d34fc106ba439b0f757c2946ca8144 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 22 Apr 2026 22:01:53 -0400 Subject: [PATCH 4/6] fix(atlas): pass WithAllowDirty(true) to all NewExecutor calls MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The testharness creates an isolated schema (wfm_test_*) that Atlas sees as a non-clean database state, causing the executor to refuse to run. WithAllowDirty(true) tells Atlas to proceed regardless of existing schemas in the database — correct for tests and for users running migrations against non-empty Postgres instances. Co-Authored-By: Claude Sonnet 4.6 --- internal/atlas/driver.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/internal/atlas/driver.go b/internal/atlas/driver.go index a5cc2f7..5ce800a 100644 --- a/internal/atlas/driver.go +++ b/internal/atlas/driver.go @@ -42,7 +42,7 @@ func (d *Driver) Up(ctx context.Context, req interfaces.MigrationRequest) (inter defer cleanup() _ = db - ex, err := atlmigrate.NewExecutor(drv, dir, rrw) + ex, err := atlmigrate.NewExecutor(drv, dir, rrw, atlmigrate.WithAllowDirty(true)) if err != nil { return interfaces.MigrationResult{}, fmt.Errorf("atlas: executor: %w", err) } @@ -147,7 +147,7 @@ func (d *Driver) Status(ctx context.Context, req interfaces.MigrationRequest) (i _ = db _ = drv - ex, err := atlmigrate.NewExecutor(drv, dir, rrw) + ex, err := atlmigrate.NewExecutor(drv, dir, rrw, atlmigrate.WithAllowDirty(true)) if err != nil { return interfaces.MigrationStatus{}, fmt.Errorf("atlas status: executor: %w", err) } @@ -205,7 +205,7 @@ func (d *Driver) Goto(ctx context.Context, req interfaces.MigrationRequest, targ // Determine direction. if target > current { // Migrate up to target. - ex, err := atlmigrate.NewExecutor(drv, dir, rrw) + ex, err := atlmigrate.NewExecutor(drv, dir, rrw, atlmigrate.WithAllowDirty(true)) if err != nil { return interfaces.MigrationResult{}, fmt.Errorf("atlas goto: executor: %w", err) } From bc8ac0b919c30de0a4ad8a722675092f00d8eb06 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 22 Apr 2026 22:12:46 -0400 Subject: [PATCH 5/6] fix(atlas): store .down.sql files in down/ subdir to prevent Atlas panic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Atlas scans all *.sql files in the migration directory. Placing .down.sql files in the same directory as up migrations causes duplicate version numbers (e.g. both 00001_create_users.sql and 00001_create_users.down.sql have version 00001). When Atlas processes the second file for an already-seen version it reads the stored revision (Applied=1) and tries to validate PartialHashes[0], which is nil in our implementation — causing a panic at migrate.go:855. Fix: require down files to live in /down/.down.sql. Update runDown() to check the preferred path first, with a legacy fallback to the flat layout. Update makeAtlasDir() in tests to write down files to the new subdirectory. Drop the now-unused "io" import. Co-Authored-By: Claude Sonnet 4.6 --- internal/atlas/driver.go | 43 +++++++++++++++++++++++------------ internal/atlas/driver_test.go | 22 +++++++++++++----- 2 files changed, 45 insertions(+), 20 deletions(-) diff --git a/internal/atlas/driver.go b/internal/atlas/driver.go index 5ce800a..2e154d0 100644 --- a/internal/atlas/driver.go +++ b/internal/atlas/driver.go @@ -9,7 +9,8 @@ import ( "database/sql" "errors" "fmt" - "io" + "os" + "path/filepath" "time" atlmigrate "ariga.io/atlas/sql/migrate" @@ -304,9 +305,16 @@ func openDir(path string) (*atlmigrate.LocalDir, error) { } // runDown executes the .down.sql file paired with a versioned Atlas migration. -// Files are expected to follow the naming: _.down.sql. +// Down files must live in a "down/" subdirectory alongside the migration directory +// to avoid Atlas treating them as up migrations (Atlas scans all *.sql files in +// the top-level directory). Expected layout: +// +// /00001_create_users.sql +// /down/00001_create_users.down.sql +// +// Falls back to /.down.sql for legacy setups. func runDown(ctx context.Context, db *sql.DB, dir, version string) error { - // Enumerate .down.sql files matching the version prefix. + // Enumerate up-migration files to find the base name for the given version. localDir, err := atlmigrate.NewLocalDir(dir) if err != nil { return err @@ -317,25 +325,32 @@ func runDown(ctx context.Context, db *sql.DB, dir, version string) error { } for _, f := range files { if f.Version() == version { - // Look for a paired .down.sql file. downName := downFilename(f.Name()) if downName == "" { return fmt.Errorf("atlas: no .down.sql file for version %s", version) } - fh, ferr := localDir.Open(downName) - if ferr != nil { - return fmt.Errorf("atlas: open %s: %w", downName, ferr) + // Preferred location: /down/ + preferredPath := filepath.Join(dir, "down", downName) + if buf, readErr := os.ReadFile(preferredPath); readErr == nil { + if len(buf) == 0 { + return fmt.Errorf("atlas: empty .down.sql for version %s", version) + } + _, err = db.ExecContext(ctx, string(buf)) + return err } - defer fh.Close() - buf, rerr := io.ReadAll(fh) - if rerr != nil { - return fmt.Errorf("atlas: read %s: %w", downName, rerr) + // Legacy fallback: / (same directory as up migrations). + // Note: storing .down.sql files alongside .sql files will cause Atlas + // to include them in its file list and may lead to duplicate-version + // errors. Use the "down/" subdirectory layout instead. + legacyPath := filepath.Join(dir, downName) + buf, readErr := os.ReadFile(legacyPath) + if readErr != nil { + return fmt.Errorf("atlas: open %s: %w", downName, readErr) } - downSQL := string(buf) - if downSQL == "" { + if len(buf) == 0 { return fmt.Errorf("atlas: empty .down.sql for version %s", version) } - _, err = db.ExecContext(ctx, downSQL) + _, err = db.ExecContext(ctx, string(buf)) return err } } diff --git a/internal/atlas/driver_test.go b/internal/atlas/driver_test.go index 1f2b4ac..a13d03a 100644 --- a/internal/atlas/driver_test.go +++ b/internal/atlas/driver_test.go @@ -85,27 +85,37 @@ func TestAtlasDriver_UpDownStatus(t *testing.T) { // makeAtlasDir creates a temporary Atlas migration directory with two versioned // SQL migrations and their down counterparts, plus a valid atlas.sum file. +// +// Down files are placed in a "down/" subdirectory so Atlas does not pick them +// up as additional up migrations (Atlas scans all *.sql files in the top-level +// directory, causing duplicate-version panics when .down.sql files share the +// same version prefix as their corresponding up migration). func makeAtlasDir(t *testing.T) string { t.Helper() dir := t.TempDir() + downDir := dir + "/down" + if err := os.MkdirAll(downDir, 0o755); err != nil { + t.Fatal(err) + } up1 := []byte("CREATE TABLE users (id SERIAL PRIMARY KEY, name TEXT NOT NULL);\n") dn1 := []byte("DROP TABLE IF EXISTS users;\n") up2 := []byte("CREATE TABLE posts (id SERIAL PRIMARY KEY, title TEXT NOT NULL);\n") dn2 := []byte("DROP TABLE IF EXISTS posts;\n") - write := func(name string, data []byte) { + write := func(path string, data []byte) { t.Helper() - if err := os.WriteFile(dir+"/"+name, data, 0o644); err != nil { + if err := os.WriteFile(path, data, 0o644); err != nil { t.Fatal(err) } } - write("00001_create_users.sql", up1) - write("00001_create_users.down.sql", dn1) - write("00002_create_posts.sql", up2) - write("00002_create_posts.down.sql", dn2) + write(dir+"/00001_create_users.sql", up1) + write(dir+"/00002_create_posts.sql", up2) + write(downDir+"/00001_create_users.down.sql", dn1) + write(downDir+"/00002_create_posts.down.sql", dn2) // Generate atlas.sum so the executor validates successfully. + // Only the top-level .sql files (up migrations) are included in the sum. localDir, err := atlmigrate.NewLocalDir(dir) if err != nil { t.Fatalf("NewLocalDir: %v", err) From dcc679cd3af7c8ba183232616506bcbf8550f244 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 22 Apr 2026 22:28:16 -0400 Subject: [PATCH 6/6] feat(cli): add test + tenant-ensure subcommands (Tasks 29+30) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit pkg/conformance/runner.go — CLI-friendly Runner (no *testing.T) with Setup() for --keep-alive and Run() for full-cycle + checkpoint cases. pkg/cli/test.go — `migrate test` cobra command: drives the conformance Runner against a user-supplied DSN + source-dir. --keep-alive applies all migrations and prints DSN for integration test setup. pkg/cli/tenant_ensure.go — `migrate tenant-ensure` cobra command: CREATE SCHEMA IF NOT EXISTS via pgx, idempotent pre-deploy step for multi-tenant environments. pkg/cli/root.go — register both new commands in NewRoot(). plugin.json — add test + tenant-ensure to cliCommands subcommands. Co-Authored-By: Claude Sonnet 4.6 --- pkg/cli/root.go | 2 + pkg/cli/tenant_ensure.go | 53 +++++++++++++ pkg/cli/test.go | 64 ++++++++++++++++ pkg/conformance/runner.go | 151 ++++++++++++++++++++++++++++++++++++++ plugin.json | 4 +- 5 files changed, 273 insertions(+), 1 deletion(-) create mode 100644 pkg/cli/tenant_ensure.go create mode 100644 pkg/cli/test.go create mode 100644 pkg/conformance/runner.go diff --git a/pkg/cli/root.go b/pkg/cli/root.go index 862f3f5..bdef8ed 100644 --- a/pkg/cli/root.go +++ b/pkg/cli/root.go @@ -47,6 +47,8 @@ func NewRoot() *cobra.Command { newStatusCmd(), newGotoCmd(), newLintCmd(), + newTestCmd(), + newTenantEnsureCmd(), ) return root } diff --git a/pkg/cli/tenant_ensure.go b/pkg/cli/tenant_ensure.go new file mode 100644 index 0000000..2b86c62 --- /dev/null +++ b/pkg/cli/tenant_ensure.go @@ -0,0 +1,53 @@ +package cli + +import ( + "context" + "database/sql" + "fmt" + "os" + + "github.com/spf13/cobra" + + _ "github.com/jackc/pgx/v5/stdlib" +) + +func newTenantEnsureCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "tenant-ensure", + Short: "Ensure a tenant schema exists in the database", + Long: `Creates the named PostgreSQL schema if it does not already exist. +Idempotent — safe to run on every deployment before running migrations. +Used as a pre-deploy step in multi-tenant environments where each tenant +has its own schema.`, + RunE: func(cmd *cobra.Command, _ []string) error { + dsn, _ := cmd.Flags().GetString("dsn") + if dsn == "" { + dsn = os.Getenv("DATABASE_URL") + } + if dsn == "" { + return fmt.Errorf("no DSN: set --dsn or DATABASE_URL env var") + } + schema, _ := cmd.Flags().GetString("schema") + if schema == "" { + return fmt.Errorf("--schema is required") + } + + db, err := sql.Open("pgx", dsn) + if err != nil { + return fmt.Errorf("tenant-ensure: open db: %w", err) + } + defer db.Close() + + if _, err = db.ExecContext(context.Background(), + fmt.Sprintf(`CREATE SCHEMA IF NOT EXISTS %q`, schema)); err != nil { + return fmt.Errorf("tenant-ensure: create schema %q: %w", schema, err) + } + fmt.Printf("Schema %q ensured.\n", schema) + return nil + }, + } + cmd.Flags().String("dsn", "", "Database connection string (overrides DATABASE_URL)") + cmd.Flags().String("schema", "", "Tenant schema name to create if absent (required)") + _ = cmd.MarkFlagRequired("schema") + return cmd +} diff --git a/pkg/cli/test.go b/pkg/cli/test.go new file mode 100644 index 0000000..4e50fe5 --- /dev/null +++ b/pkg/cli/test.go @@ -0,0 +1,64 @@ +package cli + +import ( + "context" + "fmt" + "os" + + "github.com/spf13/cobra" + + "github.com/GoCodeAlone/workflow-plugin-migrations/pkg/conformance" +) + +func newTestCmd() *cobra.Command { + cmd := &cobra.Command{ + Use: "test", + Short: "Run full-cycle and checkpoint migration tests", + Long: `Run migration conformance tests against the configured database. + +Verifies Up/Down/Status correctness with two scenarios: + full-cycle: Up all → assert no pending → Down all → assert clean state + checkpoint: Up all → Down 1 → assert 1 pending and current changed + +Use --keep-alive to apply all migrations and print DSN on stdout for +integration test setup (the calling process then uses that DB).`, + RunE: func(cmd *cobra.Command, _ []string) error { + d, req, err := buildDriverAndRequest(cmd) + if err != nil { + return err + } + keepAlive, _ := cmd.Flags().GetBool("keep-alive") + ctx := context.Background() + runner := conformance.NewRunner(req.DSN, req.Source.Dir, d) + + if keepAlive { + if err := runner.Setup(ctx); err != nil { + return fmt.Errorf("migrate test --keep-alive: %w", err) + } + fmt.Fprintln(os.Stdout, req.DSN) + return nil + } + + results := runner.Run(ctx) + pass, fail := 0, 0 + for _, r := range results { + if r.Err != nil { + fmt.Fprintf(os.Stderr, "FAIL %s: %v\n", r.Name, r.Err) + fail++ + } else { + fmt.Printf("PASS %s\n", r.Name) + pass++ + } + } + fmt.Printf("\n%d passed, %d failed\n", pass, fail) + if fail > 0 { + return fmt.Errorf("migration tests: %d failure(s)", fail) + } + return nil + }, + } + sharedFlags(cmd) + cmd.Flags().Bool("keep-alive", false, + "Apply all migrations, print DSN on stdout, and exit 0 (for integration test setup)") + return cmd +} diff --git a/pkg/conformance/runner.go b/pkg/conformance/runner.go new file mode 100644 index 0000000..5340825 --- /dev/null +++ b/pkg/conformance/runner.go @@ -0,0 +1,151 @@ +package conformance + +import ( + "context" + "fmt" + + "github.com/GoCodeAlone/workflow/interfaces" +) + +// CaseResult holds the outcome of a single conformance case. +type CaseResult struct { + Name string + Err error +} + +// Runner executes the conformance suite without requiring *testing.T. +// It is used by the `migrate test` CLI subcommand and can be embedded in +// integration-test harnesses that need pass/fail results as Go values. +type Runner struct { + dsn string + dir string + driver interfaces.MigrationDriver +} + +// NewRunner creates a Runner for the given driver, DSN, and migration directory. +func NewRunner(dsn, dir string, d interfaces.MigrationDriver) *Runner { + return &Runner{dsn: dsn, dir: dir, driver: d} +} + +// Setup applies all pending migrations. Called by `migrate test --keep-alive` to +// prepare the database for external integration tests and print the DSN. +func (r *Runner) Setup(ctx context.Context) error { + _, err := r.driver.Up(ctx, r.req()) + return err +} + +// Run executes the full-cycle and checkpoint conformance cases sequentially. +// Each case resets the database state (via repeated Down calls) before running. +func (r *Runner) Run(ctx context.Context) []CaseResult { + cases := []struct { + name string + fn func(context.Context) error + }{ + {"FullCycle", r.runFullCycle}, + {"Checkpoint", r.runCheckpoint}, + } + + results := make([]CaseResult, 0, len(cases)) + for _, c := range cases { + r.reset(ctx) // best-effort reset before each case + err := c.fn(ctx) + results = append(results, CaseResult{Name: c.name, Err: err}) + } + return results +} + +func (r *Runner) req() interfaces.MigrationRequest { + return interfaces.MigrationRequest{ + DSN: r.dsn, + Source: interfaces.MigrationSource{Dir: r.dir}, + } +} + +// reset rolls back all applied migrations to restore a clean state. +// It ignores errors (the DB may already be clean) and caps iterations +// to prevent infinite loops. +func (r *Runner) reset(ctx context.Context) { + for i := 0; i < 100; i++ { + req := r.req() + req.Options.Steps = 1 + _, err := r.driver.Down(ctx, req) + if err != nil { + return + } + st, err := r.driver.Status(ctx, r.req()) + if err != nil || st.Current == "" || st.Current == "0" { + return + } + } +} + +// runFullCycle: Up all → assert no pending/not dirty → Down all → assert clean. +func (r *Runner) runFullCycle(ctx context.Context) error { + // Apply all migrations. + result, err := r.driver.Up(ctx, r.req()) + if err != nil { + return fmt.Errorf("full-cycle Up: %w", err) + } + if len(result.Applied) == 0 { + return fmt.Errorf("full-cycle Up: expected at least one migration applied on fresh DB") + } + + // Assert nothing pending and not dirty. + st, err := r.driver.Status(ctx, r.req()) + if err != nil { + return fmt.Errorf("full-cycle Status after Up: %w", err) + } + if len(st.Pending) > 0 { + return fmt.Errorf("full-cycle: unexpected pending migrations after Up: %v", st.Pending) + } + if st.Dirty { + return fmt.Errorf("full-cycle: dirty state after Up") + } + + // Roll back all migrations. + r.reset(ctx) + + // Assert clean state. + st, err = r.driver.Status(ctx, r.req()) + if err != nil { + return fmt.Errorf("full-cycle Status after Down all: %w", err) + } + if st.Current != "" && st.Current != "0" { + return fmt.Errorf("full-cycle: expected empty current after Down all, got %q", st.Current) + } + return nil +} + +// runCheckpoint: Up all → Down 1 → assert current changed and 1 pending. +func (r *Runner) runCheckpoint(ctx context.Context) error { + // Apply all migrations. + if _, err := r.driver.Up(ctx, r.req()); err != nil { + return fmt.Errorf("checkpoint Up: %w", err) + } + + stFull, err := r.driver.Status(ctx, r.req()) + if err != nil { + return fmt.Errorf("checkpoint Status after Up: %w", err) + } + currentBefore := stFull.Current + + // Roll back one migration. + req := r.req() + req.Options.Steps = 1 + if _, err := r.driver.Down(ctx, req); err != nil { + return fmt.Errorf("checkpoint Down(1): %w", err) + } + + // Assert current changed and exactly 1 pending. + st, err := r.driver.Status(ctx, r.req()) + if err != nil { + return fmt.Errorf("checkpoint Status after Down(1): %w", err) + } + if st.Current == currentBefore { + return fmt.Errorf("checkpoint: current version unchanged after Down(1): %q", st.Current) + } + if len(st.Pending) == 0 { + return fmt.Errorf("checkpoint: expected at least 1 pending migration after Down(1)") + } + return nil +} diff --git a/plugin.json b/plugin.json index bddb8c2..70281a1 100644 --- a/plugin.json +++ b/plugin.json @@ -34,7 +34,9 @@ {"name": "down", "description": "Roll back migrations"}, {"name": "status", "description": "Show migration status"}, {"name": "goto", "description": "Migrate to a specific version"}, - {"name": "lint", "description": "Static analysis for migration files"} + {"name": "lint", "description": "Static analysis for migration files"}, + {"name": "test", "description": "Run full-cycle and checkpoint migration tests"}, + {"name": "tenant-ensure", "description": "Ensure a tenant schema exists in the database"} ], "flags_passthrough": true }