Build every database's schema from embedded SQL files (closes #56) #519
@@ -1071,23 +1071,27 @@ beside it, or wait until there are none.
|
|||||||
|
|
||||||
4. Start the service. The container gives the directory and the restored files
|
4. Start the service. The container gives the directory and the restored files
|
||||||
to the `webhooker` user before the app starts, whoever restored them (see
|
to the `webhooker` user before the app starts, whoever restored them (see
|
||||||
[Running with Docker](#running-with-docker)). `AutoMigrate` runs against each
|
[Running with Docker](#running-with-docker)). Each restored database gets any
|
||||||
restored database as it is opened.
|
of its SQL migrations it has not had as it is opened (see
|
||||||
|
[Upgrades](#upgrades)).
|
||||||
|
|
||||||
### Upgrades
|
### Upgrades
|
||||||
|
|
||||||
**Back up before every upgrade.** Every start runs GORM `AutoMigrate`
|
**Back up before every upgrade.** Each kind of database file has its schema in
|
||||||
unconditionally against whatever files it finds:
|
SQL files built into the binary, one directory per kind under
|
||||||
|
`internal/db/migrations/`. In each, `000_migration.sql` creates the
|
||||||
|
`schema_migrations` table, which records the number of every file the database
|
||||||
|
has had, and `001_schema.sql` holds the whole schema. Opening a database runs,
|
||||||
|
in one transaction, every file in its directory that `schema_migrations` does
|
||||||
|
not list:
|
||||||
|
|
||||||
- the main database on connect — `Setting`, `User`, `APIKey`, `Webhook`,
|
- `main/` for the main database, on connect — settings, users, API keys,
|
||||||
`Entrypoint`, `Target`
|
webhooks, entrypoints and targets
|
||||||
- each event database when it is lazily opened — `Event`, `Delivery`,
|
- `events/` for each event database, when it is lazily opened — events,
|
||||||
`DeliveryResult`, `EventTotals`, `TargetTotals`, `EntrypointTotals`
|
deliveries, delivery results and their running totals
|
||||||
- each archive database on every open and reopen
|
- `archive/` for each archive database, on every open and reopen
|
||||||
|
|
||||||
There is no schema version table, no migration ledger, and no down migrations.
|
There are no down migrations.
|
||||||
Nothing in the files records which version wrote them, and no code path undoes a
|
|
||||||
migration.
|
|
||||||
|
|
||||||
Upgrade procedure:
|
Upgrade procedure:
|
||||||
|
|
||||||
@@ -1112,12 +1116,9 @@ connects to. Check the `bindAddress` field of the startup log to see what a
|
|||||||
running process bound. See [Bind address](#bind-address).
|
running process bound. See [Bind address](#bind-address).
|
||||||
|
|
||||||
**Downgrade is unsupported.** Once a newer binary has migrated the files there
|
**Downgrade is unsupported.** Once a newer binary has migrated the files there
|
||||||
is no way to move them back. `AutoMigrate` is additive — it adds tables, columns
|
is no way to move them back. The only supported way back to an older version is
|
||||||
and indexes and never drops or rewrites them — so an older binary will generally
|
restoring the pre-upgrade backup, which discards everything received since that
|
||||||
open migrated files and appear to work while writing against a schema it does
|
backup was taken.
|
||||||
not know about. The failure mode is silent divergence, not a startup error. The
|
|
||||||
only supported way back to an older version is restoring the pre-upgrade backup,
|
|
||||||
which discards everything received since that backup was taken.
|
|
||||||
|
|
||||||
### Version stamping
|
### Version stamping
|
||||||
|
|
||||||
@@ -1182,26 +1183,6 @@ at rest and restrict who can read them.
|
|||||||
- `events-{uuid}.db` and `archive-….db` hold the **full payload body and
|
- `events-{uuid}.db` and `archive-….db` hold the **full payload body and
|
||||||
headers** of every event as received, including whatever the sending service
|
headers** of every event as received, including whatever the sending service
|
||||||
put in them — tokens, signatures, personal data.
|
put in them — tokens, signatures, personal data.
|
||||||
- Event databases written before
|
|
||||||
[issue #206](https://git.eeqj.de/sneak/webhooker/issues/206) was fixed **also
|
|
||||||
contain target credentials**: a GORM association upsert on the delivery and
|
|
||||||
retry write path copied `targets` rows, `config` included, into the
|
|
||||||
per-webhook database. For a Slack target the `webhookUrl` _is_ the bearer
|
|
||||||
credential, and an `http` target's URL can embed userinfo. This version never
|
|
||||||
writes those rows; the first time it opens such a file it deletes them and
|
|
||||||
vacuums the file, which removes the credential bytes rather than only
|
|
||||||
unlinking the rows. Deleting alone would not: the bytes stay readable in the
|
|
||||||
file's free pages until it is rewritten. The sweep is recorded in the file's
|
|
||||||
`user_version` only once the vacuum returns, so a sweep that fails or is
|
|
||||||
interrupted fails the open and is retried on the next one, and a file this
|
|
||||||
version has opened without error holds no leaked rows and no recoverable bytes
|
|
||||||
from them. On upgrade this rewrites each existing `events-{uuid}.db` once, on
|
|
||||||
its first open. Two cases still hand over live delivery destinations: a backup
|
|
||||||
taken from an older build, and a backup of a file this version has not yet
|
|
||||||
opened successfully. Copies already made stay affected — the sweep only
|
|
||||||
rewrites the file it opens, and freed blocks may persist in filesystem
|
|
||||||
snapshots and on the underlying storage. Rotate any target credential that was
|
|
||||||
in a backup you cannot account for.
|
|
||||||
- `webhooker.db` stores target config **unencrypted**, tracked at
|
- `webhooker.db` stores target config **unencrypted**, tracked at
|
||||||
[issue #212](https://git.eeqj.de/sneak/webhooker/issues/212), next to the
|
[issue #212](https://git.eeqj.de/sneak/webhooker/issues/212), next to the
|
||||||
session encryption key and the Argon2id password hashes.
|
session encryption key and the Argon2id password hashes.
|
||||||
@@ -1919,8 +1900,9 @@ Deliveries still `pending` or `retrying` count in neither.
|
|||||||
|
|
||||||
#### Event-tier indexes
|
#### Event-tier indexes
|
||||||
|
|
||||||
These indexes on the per-webhook event databases are declared in the model tags,
|
These indexes on the per-webhook event databases are created by
|
||||||
so `AutoMigrate` creates them on a fresh database:
|
`internal/db/migrations/events/001_schema.sql`, and declared in the model tags
|
||||||
|
to match:
|
||||||
|
|
||||||
| Table | Columns | Serves |
|
| Table | Columns | Serves |
|
||||||
| ------------------ | ------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
| ------------------ | ------------------------------------------------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||||
@@ -1984,8 +1966,9 @@ and application state:
|
|||||||
- **Targets** — delivery destination configurations
|
- **Targets** — delivery destination configurations
|
||||||
- **APIKeys** — programmatic access credentials
|
- **APIKeys** — programmatic access credentials
|
||||||
|
|
||||||
On first startup the main database is auto-migrated, a session encryption key is
|
On first startup the main database is created from its SQL files (see
|
||||||
generated and stored, and an `admin` user is created.
|
[Upgrades](#upgrades)), a session encryption key is generated and stored, and an
|
||||||
|
`admin` user is created.
|
||||||
|
|
||||||
**Per-Webhook Event Databases** (`{DATA_DIR}/events-{webhook_uuid}.db`) — each
|
**Per-Webhook Event Databases** (`{DATA_DIR}/events-{webhook_uuid}.db`) — each
|
||||||
webhook gets its own dedicated SQLite file containing:
|
webhook gets its own dedicated SQLite file containing:
|
||||||
@@ -1997,9 +1980,10 @@ webhook gets its own dedicated SQLite file containing:
|
|||||||
the above, the deliveries per target, and each entrypoint's last event, kept
|
the above, the deliveries per target, and each entrypoint's last event, kept
|
||||||
through retention
|
through retention
|
||||||
|
|
||||||
Per-webhook databases are created automatically when a webhook is created. They
|
Per-webhook databases are created automatically when a webhook is created, from
|
||||||
are managed by the `WebhookDBManager` component, which handles connection
|
the SQL files in `internal/db/migrations/events/`. They hold no copy of the main
|
||||||
pooling, lazy opening, migrations, and cleanup.
|
database's tables. They are managed by the `WebhookDBManager` component, which
|
||||||
|
handles connection pooling, lazy opening, migrations, and cleanup.
|
||||||
|
|
||||||
A per-webhook database that is missing or zero-length later means its webhook's
|
A per-webhook database that is missing or zero-length later means its webhook's
|
||||||
events and pending deliveries are gone. The next time it is opened, an empty one
|
events and pending deliveries are gone. The next time it is opened, an empty one
|
||||||
@@ -2083,14 +2067,15 @@ the service again.
|
|||||||
After each write the archive handle is closed and reopened, debounced to at most
|
After each write the archive handle is closed and reopened, debounced to at most
|
||||||
once per second, so an operator can move the archive file away for offline
|
once per second, so an operator can move the archive file away for offline
|
||||||
archiving without stopping the service; a moved or removed archive file is
|
archiving without stopping the service; a moved or removed archive file is
|
||||||
recreated automatically on the next write. A zero-length archive file is written
|
recreated automatically on the next write, from the SQL files in
|
||||||
to as a new archive: SQLite opens it as an empty database, so it holds nothing
|
`internal/db/migrations/archive/`. A zero-length archive file is written to as a
|
||||||
to lose. An optional `expiry` in the target's config JSON (e.g.
|
new archive: SQLite opens it as an empty database, so it holds nothing to lose.
|
||||||
`{"expiry":"720h"}`) is validated when the target is created — the default
|
An optional `expiry` in the target's config JSON (e.g. `{"expiry":"720h"}`) is
|
||||||
(unset or the literal `never`) keeps rows forever — and rows older than the
|
validated when the target is created — the default (unset or the literal
|
||||||
expiry are pruned each time the archive is (re)opened. An archive write failure
|
`never`) keeps rows forever — and rows older than the expiry are pruned each
|
||||||
is never silent success: the delivery records a failed attempt with the error
|
time the archive is (re)opened. An archive write failure is never silent
|
||||||
and is marked failed.
|
success: the delivery records a failed attempt with the error and is marked
|
||||||
|
failed.
|
||||||
|
|
||||||
Because reopens only happen on writes, an archive whose target has stopped
|
Because reopens only happen on writes, an archive whose target has stopped
|
||||||
receiving events would never be pruned. A background **archive sweeper** closes
|
receiving events would never be pruned. A background **archive sweeper** closes
|
||||||
@@ -3019,7 +3004,6 @@ webhooker/
|
|||||||
│ ├── database/
|
│ ├── database/
|
||||||
│ │ ├── base_model.go # BaseModel with UUID primary keys
|
│ │ ├── base_model.go # BaseModel with UUID primary keys
|
||||||
│ │ ├── database.go # GORM connection, migrations, admin seed
|
│ │ ├── database.go # GORM connection, migrations, admin seed
|
||||||
│ │ ├── models.go # AutoMigrate for config-tier models
|
|
||||||
│ │ ├── model_setting.go # Setting entity (key-value app config)
|
│ │ ├── model_setting.go # Setting entity (key-value app config)
|
||||||
│ │ ├── model_user.go # User entity
|
│ │ ├── model_user.go # User entity
|
||||||
│ │ ├── model_webhook.go # Webhook entity
|
│ │ ├── model_webhook.go # Webhook entity
|
||||||
@@ -3036,6 +3020,11 @@ webhooker/
|
|||||||
│ │ └── databasetest/ # Test support: a WebhookDBManager for tests in other packages
|
│ │ └── databasetest/ # Test support: a WebhookDBManager for tests in other packages
|
||||||
│ ├── datadir/
|
│ ├── datadir/
|
||||||
│ │ └── lock.go # Exclusive advisory lock on DATA_DIR (one instance)
|
│ │ └── lock.go # Exclusive advisory lock on DATA_DIR (one instance)
|
||||||
|
│ ├── db/migrations/
|
||||||
|
│ │ ├── migrations.go # Applies a database file's SQL files when it is opened
|
||||||
|
│ │ ├── main/ # 000_migration.sql and 001_schema.sql for webhooker.db
|
||||||
|
│ │ ├── events/ # The same for each events-{uuid}.db
|
||||||
|
│ │ └── archive/ # The same for each archive file
|
||||||
│ ├── globals/
|
│ ├── globals/
|
||||||
│ │ └── globals.go # Build-time variables (appname, version, arch)
|
│ │ └── globals.go # Build-time variables (appname, version, arch)
|
||||||
│ ├── gormlog/
|
│ ├── gormlog/
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ import (
|
|||||||
"sneak.berlin/go/webhooker/internal/banner"
|
"sneak.berlin/go/webhooker/internal/banner"
|
||||||
"sneak.berlin/go/webhooker/internal/config"
|
"sneak.berlin/go/webhooker/internal/config"
|
||||||
"sneak.berlin/go/webhooker/internal/datadir"
|
"sneak.berlin/go/webhooker/internal/datadir"
|
||||||
|
"sneak.berlin/go/webhooker/internal/db/migrations"
|
||||||
"sneak.berlin/go/webhooker/internal/gormlog"
|
"sneak.berlin/go/webhooker/internal/gormlog"
|
||||||
"sneak.berlin/go/webhooker/internal/logger"
|
"sneak.berlin/go/webhooker/internal/logger"
|
||||||
)
|
)
|
||||||
@@ -249,12 +250,21 @@ func (d *Database) connectTo(dataDir string) error {
|
|||||||
return fmt.Errorf("migrating %s: %w", dbPath, err)
|
return fmt.Errorf("migrating %s: %w", dbPath, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
err = parseModels(
|
||||||
|
db, &Setting{}, &User{}, &APIKey{},
|
||||||
|
&Webhook{}, &Entrypoint{}, &Target{},
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("connecting to %s: %w", dbPath, err)
|
||||||
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// migrate applies the main database's SQL files. See
|
||||||
|
// internal/db/migrations.
|
||||||
func (d *Database) migrate() error {
|
func (d *Database) migrate() error {
|
||||||
// Run GORM auto-migrations
|
err := migrations.Apply(d.db, migrations.Main)
|
||||||
err := d.Migrate()
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
d.log.Error(
|
d.log.Error(
|
||||||
"failed to run database migrations",
|
"failed to run database migrations",
|
||||||
@@ -269,6 +279,21 @@ func (d *Database) migrate() error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// parseModels has GORM parse each model for db before the handle is
|
||||||
|
// shared. GORM parses a model the first time a handle uses it, and two
|
||||||
|
// goroutines doing that at once for models that refer to each other,
|
||||||
|
// as Webhook and Target do, race inside GORM.
|
||||||
|
func parseModels(db *gorm.DB, models ...any) error {
|
||||||
|
for _, model := range models {
|
||||||
|
err := (&gorm.Statement{DB: db}).Parse(model)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("parsing model %T: %w", model, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// ensureAdminUser creates the bootstrap admin account when the
|
// ensureAdminUser creates the bootstrap admin account when the
|
||||||
// deployment has no users at all.
|
// deployment has no users at all.
|
||||||
func (d *Database) ensureAdminUser() error {
|
func (d *Database) ensureAdminUser() error {
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ package database
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
|
||||||
|
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
"gorm.io/gorm/clause"
|
"gorm.io/gorm/clause"
|
||||||
@@ -18,14 +17,14 @@ const omitAssociationsCallback = "webhooker:omit_associations"
|
|||||||
//
|
//
|
||||||
// A per-webhook database holds the event tier only, but Delivery
|
// A per-webhook database holds the event tier only, but Delivery
|
||||||
// declares belongs-to Event and Target and the delivery engine fills
|
// declares belongs-to Event and Target and the delivery engine fills
|
||||||
// both in memory before writing. Without this guard GORM upserts
|
// both in memory before writing. Without this guard GORM upserts those
|
||||||
// those parent rows here on the delivery and retry write paths,
|
// parent rows here on the delivery and retry write paths, and the
|
||||||
// copying targets.config, which holds destination URLs and bearer
|
// target, whose config holds destination URLs and bearer credentials,
|
||||||
// credentials, into the file most likely to be backed up or handed
|
// goes to a targets table this file does not have, which fails the
|
||||||
// to someone else. Registering the guard on the connection covers
|
// write. Registering the guard on the connection
|
||||||
// every write path, including writes inside a transaction and write
|
// covers every write path, including writes inside a transaction and
|
||||||
// paths added later. Every event-tier row this file holds is written
|
// write paths added later. Every event-tier row this file holds is
|
||||||
// explicitly, so nothing depends on the automatic save.
|
// written explicitly, so nothing depends on the automatic save.
|
||||||
func omitAssociations(db *gorm.DB) error {
|
func omitAssociations(db *gorm.DB) error {
|
||||||
omit := func(tx *gorm.DB) {
|
omit := func(tx *gorm.DB) {
|
||||||
tx.Statement.Omits = append(
|
tx.Statement.Omits = append(
|
||||||
@@ -53,107 +52,3 @@ func omitAssociations(db *gorm.DB) error {
|
|||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// eventDBSweptVersion is the PRAGMA user_version purgeTargetRows
|
|
||||||
// stamps into a per-webhook database once it has removed any leaked
|
|
||||||
// target rows *and* the VACUUM that removes their bytes has returned.
|
|
||||||
// Nothing else in the tree uses user_version, so 0 means "not swept
|
|
||||||
// by this build".
|
|
||||||
//
|
|
||||||
// The stamp, not the DELETE, is what records that a file is done. A
|
|
||||||
// DELETE commits on its own, so a sweep that is interrupted or whose
|
|
||||||
// VACUUM fails leaves a file whose rows are gone but whose credential
|
|
||||||
// bytes are still in the free pages -- indistinguishable, by row
|
|
||||||
// count, from a file that never leaked. Both leave the stamp unset,
|
|
||||||
// so the next open sweeps again.
|
|
||||||
const eventDBSweptVersion = 1
|
|
||||||
|
|
||||||
// purgeTargetRows deletes target rows that an earlier build's
|
|
||||||
// association upsert wrote into a per-webhook database, and rewrites
|
|
||||||
// the file so their bytes are gone with them. AutoMigrate creates a
|
|
||||||
// targets table in every one of these files because Delivery declares
|
|
||||||
// a belongs-to Target, but nothing in the event tier may put rows in
|
|
||||||
// it. The rows it did put there are junk, not history: they carry an
|
|
||||||
// empty webhook_id, and delivery rows resolve their target against
|
|
||||||
// the main database, so nothing here refers to them.
|
|
||||||
//
|
|
||||||
// The DELETE only unlinks the rows: modernc.org/sqlite leaves
|
|
||||||
// secure_delete at SQLite's default of off, so the credential bytes
|
|
||||||
// stay readable in the file's free pages and a backup of a swept file
|
|
||||||
// would still hand them over. VACUUM rewrites the file without them.
|
|
||||||
//
|
|
||||||
// This runs before every migration and is gated on
|
|
||||||
// eventDBSweptVersion, so a file pays for the rewrite once, on the
|
|
||||||
// first open that finds it unstamped, and every open after that is a
|
|
||||||
// PRAGMA read. A file this build created is stamped before its
|
|
||||||
// targets table exists, so it never vacuums at all. A failure here
|
|
||||||
// fails the open with the stamp left unset, so the sweep is retried
|
|
||||||
// rather than skipped -- a webhook whose file cannot be swept stays
|
|
||||||
// unusable instead of quietly serving from a file that still holds
|
|
||||||
// recoverable credentials.
|
|
||||||
func purgeTargetRows(
|
|
||||||
db *gorm.DB, log *slog.Logger, webhookID string,
|
|
||||||
) error {
|
|
||||||
var version int
|
|
||||||
|
|
||||||
// Row().Scan, not (*gorm.DB).Scan: see internal/gormlog.
|
|
||||||
err := db.Raw("PRAGMA user_version").Row().Scan(&version)
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf(
|
|
||||||
"reading sweep marker of webhook database %s: %w",
|
|
||||||
webhookID, err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
if version >= eventDBSweptVersion {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
var purged int64
|
|
||||||
|
|
||||||
if db.Migrator().HasTable("targets") {
|
|
||||||
res := db.Exec("DELETE FROM targets")
|
|
||||||
if res.Error != nil {
|
|
||||||
return fmt.Errorf(
|
|
||||||
"purging target rows from webhook database %s: %w",
|
|
||||||
webhookID, res.Error,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
purged = res.RowsAffected
|
|
||||||
|
|
||||||
// Unconditional: a zero row count here does not mean there is
|
|
||||||
// nothing to remove, only that no *live* row is left. See
|
|
||||||
// eventDBSweptVersion.
|
|
||||||
err = db.Exec("VACUUM").Error
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf(
|
|
||||||
"purged %d leaked target rows from webhook database "+
|
|
||||||
"%s but vacuuming it failed, so the deleted "+
|
|
||||||
"target credentials are still recoverable from "+
|
|
||||||
"the file; it stays marked unswept and the next "+
|
|
||||||
"open retries: %w",
|
|
||||||
purged, webhookID, err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
err = db.Exec(fmt.Sprintf(
|
|
||||||
"PRAGMA user_version = %d", eventDBSweptVersion,
|
|
||||||
)).Error
|
|
||||||
if err != nil {
|
|
||||||
return fmt.Errorf(
|
|
||||||
"marking webhook database %s swept: %w", webhookID, err,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
if purged > 0 {
|
|
||||||
log.Warn(
|
|
||||||
"purged leaked target rows from per-webhook database",
|
|
||||||
"webhook_id", webhookID,
|
|
||||||
"rows", purged,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -1,392 +1,15 @@
|
|||||||
package database_test
|
package database_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
|
||||||
"database/sql"
|
|
||||||
"fmt"
|
|
||||||
"os"
|
|
||||||
"path/filepath"
|
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
_ "modernc.org/sqlite"
|
|
||||||
"sneak.berlin/go/webhooker/internal/database"
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
"sneak.berlin/go/webhooker/internal/database/databasetest"
|
"sneak.berlin/go/webhooker/internal/database/databasetest"
|
||||||
)
|
)
|
||||||
|
|
||||||
// testDataDirPerm is the mode the test data directory is created
|
|
||||||
// with.
|
|
||||||
const testDataDirPerm = 0o750
|
|
||||||
|
|
||||||
// eventDBDataDir returns a data directory that a WebhookDBManager
|
|
||||||
// can be pointed at.
|
|
||||||
func eventDBDataDir(t *testing.T) string {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
dir := filepath.Join(t.TempDir(), "events")
|
|
||||||
require.NoError(t, os.MkdirAll(dir, testDataDirPerm))
|
|
||||||
|
|
||||||
return dir
|
|
||||||
}
|
|
||||||
|
|
||||||
// openRawEventDB opens the per-webhook database file directly,
|
|
||||||
// without the manager, so a test can put a file on disk in a state
|
|
||||||
// the manager has to cope with, or inspect one afterwards.
|
|
||||||
func openRawEventDB(
|
|
||||||
t *testing.T, dataDir, webhookID string,
|
|
||||||
) *sql.DB {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
path := filepath.Join(
|
|
||||||
dataDir, fmt.Sprintf("events-%s.db", webhookID),
|
|
||||||
)
|
|
||||||
|
|
||||||
sqlDB, err := sql.Open(
|
|
||||||
"sqlite",
|
|
||||||
fmt.Sprintf("file:%s?mode=rwc", path),
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
|
||||||
|
|
||||||
return sqlDB
|
|
||||||
}
|
|
||||||
|
|
||||||
// eventDBFileBytes reads a per-webhook database file off disk, so a
|
|
||||||
// test can assert on what the file itself still holds rather than on
|
|
||||||
// what a query returns.
|
|
||||||
func eventDBFileBytes(t *testing.T, dataDir, webhookID string) []byte {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
//nolint:gosec // reads a file the test just created under t.TempDir()
|
|
||||||
raw, err := os.ReadFile(filepath.Join(
|
|
||||||
dataDir, fmt.Sprintf("events-%s.db", webhookID),
|
|
||||||
))
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
return raw
|
|
||||||
}
|
|
||||||
|
|
||||||
// eventDBUserVersion returns the PRAGMA user_version of a per-webhook
|
|
||||||
// database file, which is the marker purgeTargetRows stamps once it
|
|
||||||
// has swept and vacuumed.
|
|
||||||
func eventDBUserVersion(t *testing.T, sqlDB *sql.DB) int {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
var version int
|
|
||||||
|
|
||||||
require.NoError(t, sqlDB.QueryRowContext(
|
|
||||||
t.Context(), "PRAGMA user_version",
|
|
||||||
).Scan(&version))
|
|
||||||
|
|
||||||
return version
|
|
||||||
}
|
|
||||||
|
|
||||||
// clearEventDBSweptMarker resets the sweep marker to 0, which is what
|
|
||||||
// a file written by a build without the sweep looks like. Tests that
|
|
||||||
// seed a leaked row have to create the file through the manager to
|
|
||||||
// get the real targets table shape, and that stamps it.
|
|
||||||
func clearEventDBSweptMarker(t *testing.T, sqlDB *sql.DB) {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
_, err := sqlDB.ExecContext(t.Context(), "PRAGMA user_version = 0")
|
|
||||||
require.NoError(t, err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// countTargetRows returns the number of rows in the targets table of
|
|
||||||
// a per-webhook database file, or -1 if the table does not exist.
|
|
||||||
func countTargetRows(t *testing.T, sqlDB *sql.DB) int {
|
|
||||||
t.Helper()
|
|
||||||
|
|
||||||
var tables int
|
|
||||||
|
|
||||||
require.NoError(t, sqlDB.QueryRowContext(
|
|
||||||
t.Context(),
|
|
||||||
"SELECT count(*) FROM sqlite_master "+
|
|
||||||
"WHERE type = 'table' AND name = 'targets'",
|
|
||||||
).Scan(&tables))
|
|
||||||
|
|
||||||
if tables == 0 {
|
|
||||||
return -1
|
|
||||||
}
|
|
||||||
|
|
||||||
var rows int
|
|
||||||
|
|
||||||
require.NoError(t, sqlDB.QueryRowContext(
|
|
||||||
t.Context(), "SELECT count(*) FROM targets",
|
|
||||||
).Scan(&rows))
|
|
||||||
|
|
||||||
return rows
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOpenPurgesLeakedTargetRows covers the sweep for event
|
|
||||||
// databases written by a build that let GORM upsert target rows
|
|
||||||
// into them: opening the database clears them, and opening it again
|
|
||||||
// is a no-op.
|
|
||||||
func TestOpenPurgesLeakedTargetRows(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
dataDir := eventDBDataDir(t)
|
|
||||||
webhookID := uuid.New().String()
|
|
||||||
|
|
||||||
// Create the file the way the application does, so the targets
|
|
||||||
// table has exactly the shape AutoMigrate gives it, then write
|
|
||||||
// a leaked row into it the way the association upsert did.
|
|
||||||
initial := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err := initial.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, initial.CloseAll())
|
|
||||||
|
|
||||||
seed := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
|
|
||||||
_, err = seed.ExecContext(
|
|
||||||
t.Context(),
|
|
||||||
"INSERT INTO targets "+
|
|
||||||
"(id, webhook_id, name, type, config) "+
|
|
||||||
"VALUES (?, '', ?, ?, ?)",
|
|
||||||
uuid.New().String(),
|
|
||||||
"leaked-target",
|
|
||||||
"slack",
|
|
||||||
`{"webhookUrl":"https://hooks.example/T000/B000/secret"}`,
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Equal(t, 1, countTargetRows(t, seed))
|
|
||||||
clearEventDBSweptMarker(t, seed)
|
|
||||||
require.NoError(t, seed.Close())
|
|
||||||
|
|
||||||
mgr := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err = mgr.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, mgr.CloseAll())
|
|
||||||
|
|
||||||
check := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
assert.Zero(t, countTargetRows(t, check))
|
|
||||||
assert.Equal(
|
|
||||||
t, 1, eventDBUserVersion(t, check),
|
|
||||||
"a completed sweep must mark the file so later opens skip it",
|
|
||||||
)
|
|
||||||
require.NoError(t, check.Close())
|
|
||||||
|
|
||||||
// Idempotent: a second open leaves it at zero and does not
|
|
||||||
// error.
|
|
||||||
again := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err = again.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, again.CloseAll())
|
|
||||||
|
|
||||||
recheck := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
assert.Zero(t, countTargetRows(t, recheck))
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOpenPurgeRemovesCredentialBytes covers the sweep at the level
|
|
||||||
// that matters for a backup handed to someone else: the leaked
|
|
||||||
// credential must be gone from the raw bytes of the file, not merely
|
|
||||||
// unreachable by query. A bare DELETE unlinks the row and leaves the
|
|
||||||
// bytes readable in the free pages, so this fails without the VACUUM
|
|
||||||
// in purgeTargetRows.
|
|
||||||
func TestOpenPurgeRemovesCredentialBytes(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
dataDir := eventDBDataDir(t)
|
|
||||||
webhookID := uuid.New().String()
|
|
||||||
credential := "T00000000/B00000000/" + uuid.New().String()
|
|
||||||
|
|
||||||
initial := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err := initial.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, initial.CloseAll())
|
|
||||||
|
|
||||||
seed := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
|
|
||||||
_, err = seed.ExecContext(
|
|
||||||
t.Context(),
|
|
||||||
"INSERT INTO targets "+
|
|
||||||
"(id, webhook_id, name, type, config) "+
|
|
||||||
"VALUES (?, '', ?, ?, ?)",
|
|
||||||
uuid.New().String(),
|
|
||||||
"leaked-target",
|
|
||||||
"slack",
|
|
||||||
fmt.Sprintf(
|
|
||||||
`{"webhookUrl":"https://hooks.example/%s"}`, credential,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
clearEventDBSweptMarker(t, seed)
|
|
||||||
require.NoError(t, seed.Close())
|
|
||||||
|
|
||||||
// The seed has to be in the file for its absence later to mean
|
|
||||||
// anything.
|
|
||||||
require.True(
|
|
||||||
t,
|
|
||||||
bytes.Contains(
|
|
||||||
eventDBFileBytes(t, dataDir, webhookID),
|
|
||||||
[]byte(credential),
|
|
||||||
),
|
|
||||||
"seeded credential is not in the file, so this test proves nothing",
|
|
||||||
)
|
|
||||||
|
|
||||||
mgr := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err = mgr.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, mgr.CloseAll())
|
|
||||||
|
|
||||||
assert.NotContains(
|
|
||||||
t,
|
|
||||||
string(eventDBFileBytes(t, dataDir, webhookID)),
|
|
||||||
credential,
|
|
||||||
"leaked credential is still recoverable from the raw file",
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOpenRevacuumsAfterIncompleteSweep covers the case a row count
|
|
||||||
// cannot see: the rows are already deleted but the file was never
|
|
||||||
// vacuumed, because an earlier sweep died between the two or its
|
|
||||||
// VACUUM failed. The credential bytes are still recoverable, and the
|
|
||||||
// unset marker is the only thing that says so, so the next open must
|
|
||||||
// vacuum rather than conclude from the empty table that there is
|
|
||||||
// nothing to do.
|
|
||||||
func TestOpenRevacuumsAfterIncompleteSweep(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
dataDir := eventDBDataDir(t)
|
|
||||||
webhookID := uuid.New().String()
|
|
||||||
credential := "T00000000/B00000000/" + uuid.New().String()
|
|
||||||
|
|
||||||
initial := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err := initial.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, initial.CloseAll())
|
|
||||||
|
|
||||||
seed := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
|
|
||||||
_, err = seed.ExecContext(
|
|
||||||
t.Context(),
|
|
||||||
"INSERT INTO targets "+
|
|
||||||
"(id, webhook_id, name, type, config) "+
|
|
||||||
"VALUES (?, '', ?, ?, ?)",
|
|
||||||
uuid.New().String(),
|
|
||||||
"leaked-target",
|
|
||||||
"slack",
|
|
||||||
fmt.Sprintf(
|
|
||||||
`{"webhookUrl":"https://hooks.example/%s"}`, credential,
|
|
||||||
),
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
// Exactly the state an interrupted sweep leaves: rows gone,
|
|
||||||
// marker unset, bytes still in the free pages.
|
|
||||||
_, err = seed.ExecContext(t.Context(), "DELETE FROM targets")
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.Zero(t, countTargetRows(t, seed))
|
|
||||||
clearEventDBSweptMarker(t, seed)
|
|
||||||
require.NoError(t, seed.Close())
|
|
||||||
|
|
||||||
require.True(
|
|
||||||
t,
|
|
||||||
bytes.Contains(
|
|
||||||
eventDBFileBytes(t, dataDir, webhookID),
|
|
||||||
[]byte(credential),
|
|
||||||
),
|
|
||||||
"the deleted row's bytes must still be in the file, or this "+
|
|
||||||
"test proves nothing",
|
|
||||||
)
|
|
||||||
|
|
||||||
mgr := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err = mgr.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, mgr.CloseAll())
|
|
||||||
|
|
||||||
assert.NotContains(
|
|
||||||
t,
|
|
||||||
string(eventDBFileBytes(t, dataDir, webhookID)),
|
|
||||||
credential,
|
|
||||||
"an interrupted sweep was not retried, so the credential is "+
|
|
||||||
"still recoverable from the raw file",
|
|
||||||
)
|
|
||||||
|
|
||||||
check := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
assert.Equal(t, 1, eventDBUserVersion(t, check))
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOpenSkipsSweptDatabase covers the other half of the marker: a
|
|
||||||
// file this build created is marked without ever being vacuumed, and
|
|
||||||
// a marked file is not swept again.
|
|
||||||
func TestOpenSkipsSweptDatabase(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
dataDir := eventDBDataDir(t)
|
|
||||||
webhookID := uuid.New().String()
|
|
||||||
|
|
||||||
mgr := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err := mgr.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, mgr.CloseAll())
|
|
||||||
|
|
||||||
marked := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
assert.Equal(t, 1, eventDBUserVersion(t, marked))
|
|
||||||
|
|
||||||
// A marked file is left alone, so a row written into it survives
|
|
||||||
// a reopen. Nothing writes target rows any more; this stands in
|
|
||||||
// for the sweep having run.
|
|
||||||
_, err = marked.ExecContext(
|
|
||||||
t.Context(),
|
|
||||||
"INSERT INTO targets "+
|
|
||||||
"(id, webhook_id, name, type, config) "+
|
|
||||||
"VALUES (?, '', ?, ?, ?)",
|
|
||||||
uuid.New().String(), "sentinel", "slack", `{}`,
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, marked.Close())
|
|
||||||
|
|
||||||
again := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
_, err = again.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, again.CloseAll())
|
|
||||||
|
|
||||||
check := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
assert.Equal(
|
|
||||||
t, 1, countTargetRows(t, check),
|
|
||||||
"a marked file must not be swept again",
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestOpenSucceedsWithoutTargetsTable covers an existing event
|
|
||||||
// database that never grew a targets table. The sweep must not fail
|
|
||||||
// startup on it.
|
|
||||||
func TestOpenSucceedsWithoutTargetsTable(t *testing.T) {
|
|
||||||
t.Parallel()
|
|
||||||
|
|
||||||
dataDir := eventDBDataDir(t)
|
|
||||||
webhookID := uuid.New().String()
|
|
||||||
|
|
||||||
seed := openRawEventDB(t, dataDir, webhookID)
|
|
||||||
|
|
||||||
_, err := seed.ExecContext(
|
|
||||||
t.Context(),
|
|
||||||
"CREATE TABLE events (id text PRIMARY KEY)",
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
require.NoError(t, seed.Close())
|
|
||||||
|
|
||||||
mgr := databasetest.NewWebhookDBManager(t, dataDir)
|
|
||||||
|
|
||||||
db, err := mgr.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
assert.NotNil(t, db)
|
|
||||||
require.NoError(t, mgr.CloseAll())
|
|
||||||
}
|
|
||||||
|
|
||||||
// TestEventDBCreateOmitsAssociations covers the connection-level
|
// TestEventDBCreateOmitsAssociations covers the connection-level
|
||||||
// guard directly: a Delivery carrying its Event and Target in
|
// guard directly: a Delivery carrying its Event and Target in
|
||||||
// memory, written through the manager's handle, must store only the
|
// memory, written through the manager's handle, must store only the
|
||||||
@@ -394,10 +17,9 @@ func TestOpenSucceedsWithoutTargetsTable(t *testing.T) {
|
|||||||
func TestEventDBCreateOmitsAssociations(t *testing.T) {
|
func TestEventDBCreateOmitsAssociations(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
dataDir := eventDBDataDir(t)
|
|
||||||
webhookID := uuid.New().String()
|
webhookID := uuid.New().String()
|
||||||
|
|
||||||
mgr := databasetest.NewWebhookDBManager(t, dataDir)
|
mgr := databasetest.NewWebhookDBManager(t, t.TempDir())
|
||||||
|
|
||||||
db, err := mgr.GetDB(webhookID)
|
db, err := mgr.GetDB(webhookID)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
@@ -432,8 +54,9 @@ func TestEventDBCreateOmitsAssociations(t *testing.T) {
|
|||||||
require.NoError(t, db.Model(d).
|
require.NoError(t, db.Model(d).
|
||||||
Update("status", database.DeliveryStatusDelivered).
|
Update("status", database.DeliveryStatusDelivered).
|
||||||
Error)
|
Error)
|
||||||
require.NoError(t, mgr.CloseAll())
|
|
||||||
|
|
||||||
check := openRawEventDB(t, dataDir, webhookID)
|
// The file has no targets table, so the create and update above
|
||||||
assert.Zero(t, countTargetRows(t, check))
|
// would have failed had GORM tried to save the target.
|
||||||
|
assert.False(t, db.Migrator().HasTable(&database.Target{}))
|
||||||
|
require.NoError(t, mgr.CloseAll())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,10 +13,8 @@ import (
|
|||||||
"sneak.berlin/go/webhooker/internal/database"
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestWebhookDBManager_OpenAddsEventTierIndexes verifies that opening a
|
// TestWebhookDBManager_OpenAddsEventTierIndexes verifies that a new
|
||||||
// per-webhook database that predates these indexes creates them. It
|
// per-webhook database has these indexes.
|
||||||
// stands in for an older database file by dropping the indexes
|
|
||||||
// AutoMigrate just created, then reopening the same file.
|
|
||||||
func TestWebhookDBManager_OpenAddsEventTierIndexes(t *testing.T) {
|
func TestWebhookDBManager_OpenAddsEventTierIndexes(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
@@ -42,27 +40,9 @@ func TestWebhookDBManager_OpenAddsEventTierIndexes(t *testing.T) {
|
|||||||
db, err := mgr.GetDB(webhookID)
|
db, err := mgr.GetDB(webhookID)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
// A fresh database has them.
|
|
||||||
for _, ix := range indexes {
|
|
||||||
require.True(t, db.Migrator().HasIndex(ix.model, ix.name))
|
|
||||||
}
|
|
||||||
|
|
||||||
// Stand in for a database file created before the indexes existed.
|
|
||||||
for _, ix := range indexes {
|
|
||||||
require.NoError(t, db.Migrator().DropIndex(ix.model, ix.name))
|
|
||||||
require.False(t, db.Migrator().HasIndex(ix.model, ix.name))
|
|
||||||
}
|
|
||||||
|
|
||||||
// Drop the cached connection so the next open reopens the file and
|
|
||||||
// runs AutoMigrate against it, as a restart would.
|
|
||||||
require.NoError(t, mgr.CloseAll())
|
|
||||||
|
|
||||||
db, err = mgr.GetDB(webhookID)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
for _, ix := range indexes {
|
for _, ix := range indexes {
|
||||||
assert.True(t, db.Migrator().HasIndex(ix.model, ix.name),
|
assert.True(t, db.Migrator().HasIndex(ix.model, ix.name),
|
||||||
"opening the existing database should create %s", ix.name)
|
"a new database should have %s", ix.name)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,17 +0,0 @@
|
|||||||
package database
|
|
||||||
|
|
||||||
// Migrate runs database migrations for the main application database.
|
|
||||||
// Only configuration-tier models are stored in the main database.
|
|
||||||
// Event-tier models (Event, Delivery, DeliveryResult, EventTotals,
|
|
||||||
// TargetTotals, EntrypointTotals) live in
|
|
||||||
// per-webhook dedicated databases managed by WebhookDBManager.
|
|
||||||
func (d *Database) Migrate() error {
|
|
||||||
return d.db.AutoMigrate(
|
|
||||||
&Setting{},
|
|
||||||
&User{},
|
|
||||||
&APIKey{},
|
|
||||||
&Webhook{},
|
|
||||||
&Entrypoint{},
|
|
||||||
&Target{},
|
|
||||||
)
|
|
||||||
}
|
|
||||||
@@ -0,0 +1,168 @@
|
|||||||
|
package database_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"log/slog"
|
||||||
|
"net/http"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/google/uuid"
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"gorm.io/gorm"
|
||||||
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
|
"sneak.berlin/go/webhooker/internal/database/databasetest"
|
||||||
|
)
|
||||||
|
|
||||||
|
// withFixedNow returns db with GORM's clock stopped at a time in UTC,
|
||||||
|
// so that the created and updated times it writes come back from
|
||||||
|
// SQLite equal to the row written, which holds them in UTC.
|
||||||
|
func withFixedNow(db *gorm.DB) (*gorm.DB, time.Time) {
|
||||||
|
at := time.Date(2026, 10, 13, 12, 30, 45, 123456789, time.UTC)
|
||||||
|
|
||||||
|
return db.Session(&gorm.Session{
|
||||||
|
NowFunc: func() time.Time { return at },
|
||||||
|
}), at
|
||||||
|
}
|
||||||
|
|
||||||
|
// newBaseModel returns a BaseModel with a new ID.
|
||||||
|
func newBaseModel() database.BaseModel {
|
||||||
|
return database.BaseModel{ID: uuid.New().String()}
|
||||||
|
}
|
||||||
|
|
||||||
|
// assertRoundTrip writes row, which must be the only row of its
|
||||||
|
// table, and reads back the same row.
|
||||||
|
func assertRoundTrip[T any](t *testing.T, db *gorm.DB, row *T) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
require.NoError(t, db.Create(row).Error)
|
||||||
|
|
||||||
|
var got T
|
||||||
|
|
||||||
|
require.NoError(t, db.First(&got).Error)
|
||||||
|
assert.Equal(t, *row, got)
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestMainSchemaHoldsEveryModel writes a row of each main database
|
||||||
|
// model, every field set, into a database made from its SQL files.
|
||||||
|
func TestMainSchemaHoldsEveryModel(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
mainDB, err := database.Open(
|
||||||
|
t.TempDir(), slog.New(slog.DiscardHandler),
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
t.Cleanup(func() { _ = mainDB.Close() })
|
||||||
|
|
||||||
|
db, at := withFixedNow(mainDB.DB())
|
||||||
|
|
||||||
|
assertRoundTrip(t, db, &database.Setting{
|
||||||
|
Key: "session_key", Value: "value",
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.User{
|
||||||
|
BaseModel: newBaseModel(),
|
||||||
|
Username: "user",
|
||||||
|
Password: "password hash",
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.APIKey{
|
||||||
|
BaseModel: newBaseModel(),
|
||||||
|
UserID: uuid.New().String(),
|
||||||
|
Key: "key",
|
||||||
|
Description: "API key description",
|
||||||
|
LastUsedAt: &at,
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.Webhook{
|
||||||
|
BaseModel: newBaseModel(),
|
||||||
|
UserID: uuid.New().String(),
|
||||||
|
Name: "webhook",
|
||||||
|
Description: "webhook description",
|
||||||
|
RetentionDays: 7,
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.Entrypoint{
|
||||||
|
BaseModel: newBaseModel(),
|
||||||
|
WebhookID: uuid.New().String(),
|
||||||
|
Path: uuid.New().String(),
|
||||||
|
Description: "entrypoint description",
|
||||||
|
Active: true,
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.Target{
|
||||||
|
BaseModel: newBaseModel(),
|
||||||
|
WebhookID: uuid.New().String(),
|
||||||
|
Name: "target",
|
||||||
|
Type: database.TargetTypeHTTP,
|
||||||
|
Active: true,
|
||||||
|
Config: `{"url":"https://example.com/"}`,
|
||||||
|
MaxRetries: 3,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestEventSchemaHoldsEveryModel does the same for a webhook's event
|
||||||
|
// database, which the manager makes from its SQL files.
|
||||||
|
func TestEventSchemaHoldsEveryModel(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
mgr := databasetest.NewWebhookDBManager(t, t.TempDir())
|
||||||
|
|
||||||
|
t.Cleanup(func() { _ = mgr.CloseAll() })
|
||||||
|
|
||||||
|
events, err := mgr.GetDB(uuid.New().String())
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
db, at := withFixedNow(events)
|
||||||
|
source := uuid.New().String()
|
||||||
|
|
||||||
|
assertRoundTrip(t, db, &database.Event{
|
||||||
|
BaseModel: newBaseModel(),
|
||||||
|
WebhookID: uuid.New().String(),
|
||||||
|
EntrypointID: uuid.New().String(),
|
||||||
|
Method: http.MethodPost,
|
||||||
|
RawQuery: "a=1",
|
||||||
|
Headers: `{"X":"Y"}`,
|
||||||
|
Body: "body",
|
||||||
|
ContentType: "text/plain",
|
||||||
|
BodyBytes: 4,
|
||||||
|
ResubmittedFromID: &source,
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.Delivery{
|
||||||
|
BaseModel: newBaseModel(),
|
||||||
|
EventID: uuid.New().String(),
|
||||||
|
TargetID: uuid.New().String(),
|
||||||
|
Status: database.DeliveryStatusDelivered,
|
||||||
|
FinishedAt: &at,
|
||||||
|
Replay: true,
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.DeliveryResult{
|
||||||
|
BaseModel: newBaseModel(),
|
||||||
|
DeliveryID: uuid.New().String(),
|
||||||
|
AttemptNum: 2,
|
||||||
|
Success: true,
|
||||||
|
StatusCode: 200,
|
||||||
|
ResponseBody: "ok",
|
||||||
|
Error: "error",
|
||||||
|
Duration: 15,
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.TargetTotals{
|
||||||
|
TargetID: uuid.New().String(),
|
||||||
|
Deliveries: 5,
|
||||||
|
Delivered: 3,
|
||||||
|
Failed: 1,
|
||||||
|
DeliveriesRemoved: 2,
|
||||||
|
FailedRemoved: 1,
|
||||||
|
})
|
||||||
|
assertRoundTrip(t, db, &database.EntrypointTotals{
|
||||||
|
EntrypointID: uuid.New().String(), LastEventAt: at,
|
||||||
|
})
|
||||||
|
|
||||||
|
// A new event database has its one row of event totals already.
|
||||||
|
require.NoError(t, database.AddEventTotals(db, database.EventTotals{
|
||||||
|
Events: 3, EventsRemoved: 1, LastEventAt: &at,
|
||||||
|
}))
|
||||||
|
|
||||||
|
var totals []database.EventTotals
|
||||||
|
|
||||||
|
require.NoError(t, db.Find(&totals).Error)
|
||||||
|
assert.Equal(t, []database.EventTotals{{
|
||||||
|
ID: 1, Events: 3, EventsRemoved: 1, LastEventAt: &at,
|
||||||
|
}}, totals)
|
||||||
|
}
|
||||||
@@ -14,6 +14,7 @@ import (
|
|||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
"sneak.berlin/go/webhooker/internal/config"
|
"sneak.berlin/go/webhooker/internal/config"
|
||||||
"sneak.berlin/go/webhooker/internal/datadir"
|
"sneak.berlin/go/webhooker/internal/datadir"
|
||||||
|
"sneak.berlin/go/webhooker/internal/db/migrations"
|
||||||
"sneak.berlin/go/webhooker/internal/gormlog"
|
"sneak.berlin/go/webhooker/internal/gormlog"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -370,18 +371,7 @@ func (m *WebhookDBManager) openDB(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = purgeTargetRows(db, m.log, webhookID)
|
err = migrations.Apply(db, migrations.Events)
|
||||||
if err != nil {
|
|
||||||
_ = sqlDB.Close()
|
|
||||||
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Run migrations for event-tier models only
|
|
||||||
err = db.AutoMigrate(
|
|
||||||
&Event{}, &Delivery{}, &DeliveryResult{},
|
|
||||||
&EventTotals{}, &TargetTotals{}, &EntrypointTotals{},
|
|
||||||
)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
_ = sqlDB.Close()
|
_ = sqlDB.Close()
|
||||||
|
|
||||||
@@ -391,14 +381,15 @@ func (m *WebhookDBManager) openDB(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
// A new database gets its row of event totals, all zero. Target
|
err = parseModels(
|
||||||
// totals rows are created by the first delivery to each target.
|
db, &Event{}, &Delivery{}, &DeliveryResult{},
|
||||||
err = db.FirstOrCreate(&EventTotals{}).Error
|
&EventTotals{}, &TargetTotals{}, &EntrypointTotals{},
|
||||||
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
_ = sqlDB.Close()
|
_ = sqlDB.Close()
|
||||||
|
|
||||||
return nil, fmt.Errorf(
|
return nil, fmt.Errorf(
|
||||||
"creating event totals for webhook database %s: %w",
|
"connecting to webhook database %s: %w",
|
||||||
webhookID, err,
|
webhookID, err,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,9 @@
|
|||||||
|
-- Migration 000: the table recording which migrations this file has
|
||||||
|
-- had. Applied before the others; it records itself.
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||||
|
version INTEGER PRIMARY KEY,
|
||||||
|
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||||
|
);
|
||||||
|
|
||||||
|
INSERT OR IGNORE INTO schema_migrations (version) VALUES (0);
|
||||||
@@ -0,0 +1,18 @@
|
|||||||
|
-- Migration 001: the schema of a database target's archive file. The
|
||||||
|
-- archivedEvent model in internal/delivery must agree with it.
|
||||||
|
|
||||||
|
CREATE TABLE archived_events (
|
||||||
|
id integer PRIMARY KEY AUTOINCREMENT,
|
||||||
|
event_id text,
|
||||||
|
webhook_id text,
|
||||||
|
entrypoint_id text,
|
||||||
|
method text,
|
||||||
|
raw_query text,
|
||||||
|
headers text,
|
||||||
|
body text,
|
||||||
|
content_type text,
|
||||||
|
archived_at datetime
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_archived_events_event_id ON archived_events (event_id);
|
||||||
|
CREATE INDEX idx_archived_events_archived_at ON archived_events (archived_at);
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
-- Migration 000: the table recording which migrations this file has
|
||||||
|
-- had. Applied before the others; it records itself.
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||||
|
version INTEGER PRIMARY KEY,
|
||||||
|
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||||
|
);
|
||||||
|
|
||||||
|
INSERT OR IGNORE INTO schema_migrations (version) VALUES (0);
|
||||||
@@ -0,0 +1,98 @@
|
|||||||
|
-- Migration 001: the schema of a webhook's event database. The models
|
||||||
|
-- in internal/database must agree with it. The webhook, entrypoint and
|
||||||
|
-- target ids here name rows in the main database, which has those
|
||||||
|
-- tables; this file does not.
|
||||||
|
|
||||||
|
CREATE TABLE events (
|
||||||
|
id uuid,
|
||||||
|
created_at datetime,
|
||||||
|
updated_at datetime,
|
||||||
|
deleted_at datetime,
|
||||||
|
webhook_id uuid NOT NULL,
|
||||||
|
entrypoint_id uuid NOT NULL,
|
||||||
|
method text NOT NULL,
|
||||||
|
raw_query text,
|
||||||
|
headers text,
|
||||||
|
body text,
|
||||||
|
content_type text,
|
||||||
|
body_bytes integer NOT NULL,
|
||||||
|
resubmitted_from_id uuid,
|
||||||
|
PRIMARY KEY (id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_events_created_at ON events (created_at);
|
||||||
|
CREATE INDEX idx_events_deleted_at ON events (deleted_at);
|
||||||
|
CREATE INDEX idx_events_deleted_at_created_at
|
||||||
|
ON events (deleted_at, created_at);
|
||||||
|
CREATE INDEX idx_events_resubmitted_from_id
|
||||||
|
ON events (resubmitted_from_id, deleted_at);
|
||||||
|
CREATE INDEX idx_events_entrypoint_id
|
||||||
|
ON events (entrypoint_id, deleted_at, resubmitted_from_id, created_at);
|
||||||
|
|
||||||
|
CREATE TABLE deliveries (
|
||||||
|
id uuid,
|
||||||
|
created_at datetime,
|
||||||
|
updated_at datetime,
|
||||||
|
deleted_at datetime,
|
||||||
|
event_id uuid NOT NULL,
|
||||||
|
target_id uuid NOT NULL,
|
||||||
|
status text NOT NULL DEFAULT 'pending',
|
||||||
|
finished_at datetime,
|
||||||
|
replay numeric NOT NULL DEFAULT false,
|
||||||
|
PRIMARY KEY (id),
|
||||||
|
CONSTRAINT fk_events_deliveries
|
||||||
|
FOREIGN KEY (event_id) REFERENCES events (id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_deliveries_deleted_at ON deliveries (deleted_at);
|
||||||
|
CREATE INDEX idx_deliveries_event_id ON deliveries (event_id, deleted_at);
|
||||||
|
CREATE INDEX idx_deliveries_status
|
||||||
|
ON deliveries (status, deleted_at, finished_at, target_id);
|
||||||
|
|
||||||
|
CREATE TABLE delivery_results (
|
||||||
|
id uuid,
|
||||||
|
created_at datetime,
|
||||||
|
updated_at datetime,
|
||||||
|
deleted_at datetime,
|
||||||
|
delivery_id uuid NOT NULL,
|
||||||
|
attempt_num integer NOT NULL,
|
||||||
|
success numeric,
|
||||||
|
status_code integer,
|
||||||
|
response_body text,
|
||||||
|
error text,
|
||||||
|
duration integer,
|
||||||
|
PRIMARY KEY (id),
|
||||||
|
CONSTRAINT fk_deliveries_delivery_results
|
||||||
|
FOREIGN KEY (delivery_id) REFERENCES deliveries (id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_delivery_results_deleted_at
|
||||||
|
ON delivery_results (deleted_at);
|
||||||
|
CREATE INDEX idx_delivery_results_delivery_id
|
||||||
|
ON delivery_results (delivery_id, deleted_at);
|
||||||
|
|
||||||
|
CREATE TABLE event_totals (
|
||||||
|
id integer PRIMARY KEY AUTOINCREMENT,
|
||||||
|
events integer NOT NULL,
|
||||||
|
events_removed integer NOT NULL,
|
||||||
|
last_event_at datetime
|
||||||
|
);
|
||||||
|
|
||||||
|
-- The one row of event totals, all zero.
|
||||||
|
INSERT INTO event_totals (events, events_removed) VALUES (0, 0);
|
||||||
|
|
||||||
|
CREATE TABLE target_totals (
|
||||||
|
target_id uuid,
|
||||||
|
deliveries integer NOT NULL,
|
||||||
|
delivered integer NOT NULL,
|
||||||
|
failed integer NOT NULL,
|
||||||
|
deliveries_removed integer NOT NULL,
|
||||||
|
failed_removed integer NOT NULL,
|
||||||
|
PRIMARY KEY (target_id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE entrypoint_totals (
|
||||||
|
entrypoint_id uuid,
|
||||||
|
last_event_at datetime NOT NULL,
|
||||||
|
PRIMARY KEY (entrypoint_id)
|
||||||
|
);
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
-- Migration 000: the table recording which migrations this file has
|
||||||
|
-- had. Applied before the others; it records itself.
|
||||||
|
|
||||||
|
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||||||
|
version INTEGER PRIMARY KEY,
|
||||||
|
applied_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||||||
|
);
|
||||||
|
|
||||||
|
INSERT OR IGNORE INTO schema_migrations (version) VALUES (0);
|
||||||
@@ -0,0 +1,89 @@
|
|||||||
|
-- Migration 001: the main database's schema. The models in
|
||||||
|
-- internal/database must agree with it.
|
||||||
|
|
||||||
|
CREATE TABLE settings (
|
||||||
|
"key" text,
|
||||||
|
value text NOT NULL,
|
||||||
|
PRIMARY KEY ("key")
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE TABLE users (
|
||||||
|
id uuid,
|
||||||
|
created_at datetime,
|
||||||
|
updated_at datetime,
|
||||||
|
deleted_at datetime,
|
||||||
|
username text NOT NULL,
|
||||||
|
password text NOT NULL,
|
||||||
|
PRIMARY KEY (id),
|
||||||
|
CONSTRAINT chk_users_username
|
||||||
|
CHECK (length(CAST(username AS BLOB)) <= 1024)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_users_deleted_at ON users (deleted_at);
|
||||||
|
CREATE UNIQUE INDEX idx_users_username ON users (username);
|
||||||
|
|
||||||
|
CREATE TABLE api_keys (
|
||||||
|
id uuid,
|
||||||
|
created_at datetime,
|
||||||
|
updated_at datetime,
|
||||||
|
deleted_at datetime,
|
||||||
|
user_id uuid NOT NULL,
|
||||||
|
"key" text NOT NULL,
|
||||||
|
description text,
|
||||||
|
last_used_at datetime,
|
||||||
|
PRIMARY KEY (id),
|
||||||
|
CONSTRAINT fk_users_api_keys FOREIGN KEY (user_id) REFERENCES users (id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_api_keys_deleted_at ON api_keys (deleted_at);
|
||||||
|
CREATE UNIQUE INDEX idx_api_keys_key ON api_keys ("key");
|
||||||
|
|
||||||
|
CREATE TABLE webhooks (
|
||||||
|
id uuid,
|
||||||
|
created_at datetime,
|
||||||
|
updated_at datetime,
|
||||||
|
deleted_at datetime,
|
||||||
|
user_id uuid NOT NULL,
|
||||||
|
name text NOT NULL,
|
||||||
|
description text,
|
||||||
|
retention_days integer DEFAULT 30,
|
||||||
|
PRIMARY KEY (id),
|
||||||
|
CONSTRAINT fk_users_webhooks FOREIGN KEY (user_id) REFERENCES users (id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_webhooks_deleted_at ON webhooks (deleted_at);
|
||||||
|
|
||||||
|
CREATE TABLE entrypoints (
|
||||||
|
id uuid,
|
||||||
|
created_at datetime,
|
||||||
|
updated_at datetime,
|
||||||
|
deleted_at datetime,
|
||||||
|
webhook_id uuid NOT NULL,
|
||||||
|
path text NOT NULL,
|
||||||
|
description text,
|
||||||
|
active numeric DEFAULT true,
|
||||||
|
PRIMARY KEY (id),
|
||||||
|
CONSTRAINT fk_webhooks_entrypoints
|
||||||
|
FOREIGN KEY (webhook_id) REFERENCES webhooks (id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_entrypoints_deleted_at ON entrypoints (deleted_at);
|
||||||
|
CREATE UNIQUE INDEX idx_entrypoints_path ON entrypoints (path);
|
||||||
|
|
||||||
|
CREATE TABLE targets (
|
||||||
|
id uuid,
|
||||||
|
created_at datetime,
|
||||||
|
updated_at datetime,
|
||||||
|
deleted_at datetime,
|
||||||
|
webhook_id uuid NOT NULL,
|
||||||
|
name text NOT NULL,
|
||||||
|
type text NOT NULL,
|
||||||
|
active numeric DEFAULT true,
|
||||||
|
config text,
|
||||||
|
max_retries integer,
|
||||||
|
PRIMARY KEY (id),
|
||||||
|
CONSTRAINT fk_webhooks_targets
|
||||||
|
FOREIGN KEY (webhook_id) REFERENCES webhooks (id)
|
||||||
|
);
|
||||||
|
|
||||||
|
CREATE INDEX idx_targets_deleted_at ON targets (deleted_at);
|
||||||
@@ -0,0 +1,135 @@
|
|||||||
|
// Package migrations holds the schema of each kind of SQLite file
|
||||||
|
// webhooker writes, as numbered SQL files embedded in the binary, and
|
||||||
|
// applies it when a file is opened.
|
||||||
|
//
|
||||||
|
// Each kind of file has its own directory. In each, 000_migration.sql
|
||||||
|
// creates only the schema_migrations table, and 001_schema.sql holds
|
||||||
|
// the whole schema. Before 1.0 a schema change is an edit to
|
||||||
|
// 001_schema.sql, never a new file.
|
||||||
|
package migrations
|
||||||
|
|
||||||
|
import (
|
||||||
|
"embed"
|
||||||
|
"fmt"
|
||||||
|
"io/fs"
|
||||||
|
"path"
|
||||||
|
"strconv"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"gorm.io/gorm"
|
||||||
|
)
|
||||||
|
|
||||||
|
// The directories, one for each kind of database file.
|
||||||
|
const (
|
||||||
|
// Main is the main database, webhooker.db.
|
||||||
|
Main = "main"
|
||||||
|
|
||||||
|
// Events is a webhook's event database, events-{uuid}.db.
|
||||||
|
Events = "events"
|
||||||
|
|
||||||
|
// Archive is a database target's archive file.
|
||||||
|
Archive = "archive"
|
||||||
|
)
|
||||||
|
|
||||||
|
//go:embed main/*.sql events/*.sql archive/*.sql
|
||||||
|
var files embed.FS
|
||||||
|
|
||||||
|
// Apply runs each SQL file in dir, which is Main, Events or Archive,
|
||||||
|
// that the database has not had yet, in the order of the number its
|
||||||
|
// name starts with, and records that number in schema_migrations.
|
||||||
|
// 000_migration.sql creates that table and records itself, so it runs
|
||||||
|
// first, and only when the table is missing.
|
||||||
|
//
|
||||||
|
// It all runs in one transaction, so a failure leaves the file as it
|
||||||
|
// was.
|
||||||
|
func Apply(db *gorm.DB, dir string) error {
|
||||||
|
return db.Transaction(func(tx *gorm.DB) error {
|
||||||
|
var tables int
|
||||||
|
|
||||||
|
// Row().Scan, not (*gorm.DB).Scan: see internal/gormlog.
|
||||||
|
err := tx.Raw(
|
||||||
|
"SELECT count(*) FROM sqlite_master "+
|
||||||
|
"WHERE type = 'table' AND name = ?",
|
||||||
|
"schema_migrations",
|
||||||
|
).Row().Scan(&tables)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("looking for schema_migrations: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if tables == 0 {
|
||||||
|
err = run(tx, path.Join(dir, "000_migration.sql"))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// fs.Glob lists the files in name order, which is number order.
|
||||||
|
names, err := fs.Glob(files, path.Join(dir, "*.sql"))
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("listing migrations: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, name := range names {
|
||||||
|
err = applyOnce(tx, name)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
// applyOnce runs the migration file name and records its number,
|
||||||
|
// unless schema_migrations already has that number.
|
||||||
|
func applyOnce(tx *gorm.DB, name string) error {
|
||||||
|
number, _, _ := strings.Cut(path.Base(name), "_")
|
||||||
|
|
||||||
|
version, err := strconv.Atoi(number)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("numbering migration %s: %w", name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
var applied int
|
||||||
|
|
||||||
|
err = tx.Raw(
|
||||||
|
"SELECT count(*) FROM schema_migrations WHERE version = ?",
|
||||||
|
version,
|
||||||
|
).Row().Scan(&applied)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("checking migration %s: %w", name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
if applied > 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
err = run(tx, name)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
err = tx.Exec(
|
||||||
|
"INSERT INTO schema_migrations (version) VALUES (?)", version,
|
||||||
|
).Error
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("recording migration %s: %w", name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// run executes every statement in the migration file name.
|
||||||
|
func run(tx *gorm.DB, name string) error {
|
||||||
|
script, err := files.ReadFile(name)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("reading migration %s: %w", name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
err = tx.Exec(string(script)).Error
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("applying migration %s: %w", name, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
@@ -0,0 +1,57 @@
|
|||||||
|
package migrations_test
|
||||||
|
|
||||||
|
import (
|
||||||
|
"database/sql"
|
||||||
|
"log/slog"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/assert"
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
"gorm.io/driver/sqlite"
|
||||||
|
"gorm.io/gorm"
|
||||||
|
_ "modernc.org/sqlite"
|
||||||
|
"sneak.berlin/go/webhooker/internal/db/migrations"
|
||||||
|
"sneak.berlin/go/webhooker/internal/gormlog"
|
||||||
|
)
|
||||||
|
|
||||||
|
// TestApplyRunsEachFileOnce applies each kind of database file's
|
||||||
|
// migrations to a new file twice. The first time records every file.
|
||||||
|
// The second finds them all recorded and runs none, which is the only
|
||||||
|
// way it can succeed: 001_schema.sql creates tables that then exist.
|
||||||
|
func TestApplyRunsEachFileOnce(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
for _, dir := range []string{
|
||||||
|
migrations.Main, migrations.Events, migrations.Archive,
|
||||||
|
} {
|
||||||
|
t.Run(dir, func(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
sqlDB, err := sql.Open(
|
||||||
|
"sqlite", filepath.Join(t.TempDir(), "test.db"),
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
t.Cleanup(func() { _ = sqlDB.Close() })
|
||||||
|
|
||||||
|
db, err := gorm.Open(
|
||||||
|
sqlite.Dialector{Conn: sqlDB},
|
||||||
|
&gorm.Config{
|
||||||
|
Logger: gormlog.New(slog.New(slog.DiscardHandler)),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
require.NoError(t, migrations.Apply(db, dir))
|
||||||
|
require.NoError(t, migrations.Apply(db, dir))
|
||||||
|
|
||||||
|
var versions string
|
||||||
|
|
||||||
|
require.NoError(t, sqlDB.QueryRowContext(t.Context(),
|
||||||
|
"SELECT group_concat(version) FROM schema_migrations",
|
||||||
|
).Scan(&versions))
|
||||||
|
assert.Equal(t, "0,1", versions)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -22,6 +22,7 @@ import (
|
|||||||
_ "modernc.org/sqlite" // Pure Go SQLite driver.
|
_ "modernc.org/sqlite" // Pure Go SQLite driver.
|
||||||
"sneak.berlin/go/webhooker/internal/database"
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
"sneak.berlin/go/webhooker/internal/database/databasetest"
|
"sneak.berlin/go/webhooker/internal/database/databasetest"
|
||||||
|
"sneak.berlin/go/webhooker/internal/db/migrations"
|
||||||
"sneak.berlin/go/webhooker/internal/delivery"
|
"sneak.berlin/go/webhooker/internal/delivery"
|
||||||
"sneak.berlin/go/webhooker/internal/gormlog"
|
"sneak.berlin/go/webhooker/internal/gormlog"
|
||||||
)
|
)
|
||||||
@@ -172,9 +173,7 @@ func seedArchiveFile(
|
|||||||
)
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
require.NoError(
|
require.NoError(t, migrations.Apply(gdb, migrations.Archive))
|
||||||
t, gdb.AutoMigrate(&delivery.ExportArchivedEvent{}),
|
|
||||||
)
|
|
||||||
|
|
||||||
for i, at := range archivedAt {
|
for i, at := range archivedAt {
|
||||||
row := delivery.ExportArchivedEvent{
|
row := delivery.ExportArchivedEvent{
|
||||||
@@ -781,9 +780,9 @@ func TestArchiveSweep_NeverExpiryUntouched(t *testing.T) {
|
|||||||
//
|
//
|
||||||
// The spec is stronger than that — a "never" archive is skipped
|
// The spec is stronger than that — a "never" archive is skipped
|
||||||
// before any file is touched — so the archive here exists but has
|
// before any file is touched — so the archive here exists but has
|
||||||
// never been migrated. Opening it at all would run AutoMigrate
|
// never been migrated. Opening it at all would apply the archive
|
||||||
// and create the archive table, which is exactly what must not
|
// schema and create the archive table, which is exactly what must
|
||||||
// happen.
|
// not happen.
|
||||||
func TestArchiveSweep_NeverExpirySkipsBeforeOpening(
|
func TestArchiveSweep_NeverExpirySkipsBeforeOpening(
|
||||||
t *testing.T,
|
t *testing.T,
|
||||||
) {
|
) {
|
||||||
@@ -807,7 +806,7 @@ func TestArchiveSweep_NeverExpirySkipsBeforeOpening(
|
|||||||
|
|
||||||
// seedUnmigratedArchive creates an archive file that exists but
|
// seedUnmigratedArchive creates an archive file that exists but
|
||||||
// carries no archive schema, so any open of it is observable: the
|
// carries no archive schema, so any open of it is observable: the
|
||||||
// archive table appears only if something ran AutoMigrate.
|
// archive table appears only if something applied the schema.
|
||||||
func seedUnmigratedArchive(t *testing.T, path string) {
|
func seedUnmigratedArchive(t *testing.T, path string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
|
|||||||
@@ -963,29 +963,12 @@ func TestFailUnretryableRetry_WritesNoTargetRow(
|
|||||||
database.DeliveryStatusFailed,
|
database.DeliveryStatusFailed,
|
||||||
)
|
)
|
||||||
|
|
||||||
// The table exists in the per-webhook database because GORM
|
// The per-webhook database has no targets table, so writing a
|
||||||
// migrates the Delivery relation's model alongside it. It must
|
// target row there would have failed the status write above.
|
||||||
// stay empty.
|
assert.False(t, s.WebhookDB.Migrator().HasTable("targets"),
|
||||||
var targetRows int64
|
|
||||||
|
|
||||||
require.NoError(t, s.WebhookDB.
|
|
||||||
Table("targets").
|
|
||||||
Count(&targetRows).Error)
|
|
||||||
|
|
||||||
assert.Zero(t, targetRows,
|
|
||||||
"orphaned-retry terminal failure wrote a target row "+
|
"orphaned-retry terminal failure wrote a target row "+
|
||||||
"into the per-webhook event database",
|
"into the per-webhook event database",
|
||||||
)
|
)
|
||||||
|
|
||||||
var configs []string
|
|
||||||
|
|
||||||
require.NoError(t, s.WebhookDB.
|
|
||||||
Table("targets").
|
|
||||||
Pluck("config", &configs).Error)
|
|
||||||
|
|
||||||
assert.NotContains(
|
|
||||||
t, strings.Join(configs, " "), hookURL,
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestRecoverSingleRetry_UnknownTargetType(
|
func TestRecoverSingleRetry_UnknownTargetType(
|
||||||
|
|||||||
@@ -10,7 +10,6 @@ import (
|
|||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
"os"
|
"os"
|
||||||
"path/filepath"
|
|
||||||
"strconv"
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -22,12 +21,11 @@ import (
|
|||||||
"github.com/prometheus/client_golang/prometheus"
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"gorm.io/driver/sqlite"
|
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
_ "modernc.org/sqlite"
|
_ "modernc.org/sqlite"
|
||||||
"sneak.berlin/go/webhooker/internal/database"
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
|
"sneak.berlin/go/webhooker/internal/database/databasetest"
|
||||||
"sneak.berlin/go/webhooker/internal/delivery"
|
"sneak.berlin/go/webhooker/internal/delivery"
|
||||||
"sneak.berlin/go/webhooker/internal/gormlog"
|
|
||||||
"sneak.berlin/go/webhooker/internal/metrics"
|
"sneak.berlin/go/webhooker/internal/metrics"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -37,35 +35,18 @@ const testContentType = "application/json"
|
|||||||
func testWebhookDB(t *testing.T) *gorm.DB {
|
func testWebhookDB(t *testing.T) *gorm.DB {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
dbPath := filepath.Join(
|
// Opened by the manager the service uses, so these tests cannot
|
||||||
t.TempDir(), "events-test.db",
|
// pass against a schema, journal and locking settings or write
|
||||||
|
// guard production does not use.
|
||||||
|
mgr := databasetest.NewWebhookDBManagerWithLogger(
|
||||||
|
t, t.TempDir(), slog.New(slog.DiscardHandler),
|
||||||
)
|
)
|
||||||
|
|
||||||
// Opened the way the service opens a per-webhook database, so
|
t.Cleanup(func() { _ = mgr.CloseAll() })
|
||||||
// these tests cannot pass against journal and locking settings
|
|
||||||
// production does not use.
|
db, err := mgr.GetDB(uuid.New().String())
|
||||||
sqlDB, err := database.OpenSQLite(
|
|
||||||
dbPath, database.SQLiteModeCreate,
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
t.Cleanup(func() { _ = sqlDB.Close() })
|
|
||||||
|
|
||||||
db, err := gorm.Open(
|
|
||||||
sqlite.Dialector{Conn: sqlDB},
|
|
||||||
&gorm.Config{Logger: gormlog.New(slog.New(slog.DiscardHandler))},
|
|
||||||
)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
require.NoError(t, db.AutoMigrate(
|
|
||||||
&database.Event{},
|
|
||||||
&database.Delivery{},
|
|
||||||
&database.DeliveryResult{},
|
|
||||||
&database.EventTotals{},
|
|
||||||
&database.TargetTotals{},
|
|
||||||
))
|
|
||||||
require.NoError(t, db.Create(&database.EventTotals{}).Error)
|
|
||||||
|
|
||||||
return db
|
return db
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -14,6 +14,7 @@ import (
|
|||||||
"gorm.io/driver/sqlite"
|
"gorm.io/driver/sqlite"
|
||||||
"gorm.io/gorm"
|
"gorm.io/gorm"
|
||||||
"sneak.berlin/go/webhooker/internal/database"
|
"sneak.berlin/go/webhooker/internal/database"
|
||||||
|
"sneak.berlin/go/webhooker/internal/db/migrations"
|
||||||
"sneak.berlin/go/webhooker/internal/gormlog"
|
"sneak.berlin/go/webhooker/internal/gormlog"
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -335,7 +336,7 @@ func (w *archiveWriter) openMode(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
err = gdb.AutoMigrate(&archivedEvent{})
|
err = migrations.Apply(gdb, migrations.Archive)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
_ = sqlDB.Close()
|
_ = sqlDB.Close()
|
||||||
|
|
||||||
|
|||||||
@@ -312,13 +312,9 @@ func TestFailMissingTarget_WritesNoTargetRow(
|
|||||||
database.DeliveryStatusFailed,
|
database.DeliveryStatusFailed,
|
||||||
)
|
)
|
||||||
|
|
||||||
var configs []string
|
// The per-webhook database has no targets table, so writing a
|
||||||
|
// target row there would have failed the status write above.
|
||||||
require.NoError(t, s.WebhookDB.
|
assert.False(t, s.WebhookDB.Migrator().HasTable("targets"),
|
||||||
Table("targets").
|
|
||||||
Pluck("config", &configs).Error)
|
|
||||||
|
|
||||||
assert.Empty(t, configs,
|
|
||||||
"the deleted-target terminal path wrote a target row "+
|
"the deleted-target terminal path wrote a target row "+
|
||||||
"into the per-webhook event database",
|
"into the per-webhook event database",
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -135,9 +135,11 @@ func openDB(
|
|||||||
)
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
require.NoError(t, gdb.AutoMigrate(&thing{}))
|
require.NoError(t, gdb.Exec(
|
||||||
|
"CREATE TABLE things (id text PRIMARY KEY, name text)",
|
||||||
|
).Error)
|
||||||
|
|
||||||
// Migration chatter is not what any of these cases is about.
|
// Creating the table is not what any of these cases is about.
|
||||||
buf.Reset()
|
buf.Reset()
|
||||||
|
|
||||||
return gdb
|
return gdb
|
||||||
|
|||||||
@@ -115,8 +115,8 @@ func skipDir(name string) bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// walkNonTestGo parses every non-test .go file under root. It returns
|
// walkNonTestGo parses every non-test .go file under root. It returns
|
||||||
// the directories, relative to root, it parsed a file in, along with
|
// the directories, relative to root, it parsed a file in or below,
|
||||||
// every unguarded Scan it found.
|
// along with every unguarded Scan it found.
|
||||||
func walkNonTestGo(t *testing.T, root string) (map[string]bool, []string) {
|
func walkNonTestGo(t *testing.T, root string) (map[string]bool, []string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
@@ -155,7 +155,12 @@ func walkNonTestGo(t *testing.T, root string) (map[string]bool, []string) {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
walked[dir] = true
|
// The directories above count too, so that one holding
|
||||||
|
// only a nested package, as internal/db does, is not
|
||||||
|
// taken for one the walk skipped.
|
||||||
|
for ; dir != "."; dir = filepath.Dir(dir) {
|
||||||
|
walked[dir] = true
|
||||||
|
}
|
||||||
|
|
||||||
for _, pos := range unguardedScans(fset, file) {
|
for _, pos := range unguardedScans(fset, file) {
|
||||||
hits = append(hits, relPosition(root, pos))
|
hits = append(hits, relPosition(root, pos))
|
||||||
@@ -205,9 +210,9 @@ func TestGormScanIsNeverCalledOutsideTests(t *testing.T) {
|
|||||||
walked, offenders := walkNonTestGo(t, root)
|
walked, offenders := walkNonTestGo(t, root)
|
||||||
|
|
||||||
// The module's packages are static, templates, and every directory
|
// The module's packages are static, templates, and every directory
|
||||||
// directly under cmd and internal. Each holds non-test code, so one
|
// directly under cmd and internal. Each holds non-test code, itself
|
||||||
// the walk parsed nothing in was skipped, and a Scan there would
|
// or in a package below it, so one the walk parsed nothing in or
|
||||||
// pass unseen.
|
// below was skipped, and a Scan there would pass unseen.
|
||||||
packages := []string{"static", "templates"}
|
packages := []string{"static", "templates"}
|
||||||
|
|
||||||
for _, parent := range []string{"cmd", "internal"} {
|
for _, parent := range []string{"cmd", "internal"} {
|
||||||
|
|||||||
@@ -42,8 +42,11 @@ func TestDeliveryAttempts_ReadInTheTargetTypesOwnTerms(t *testing.T) {
|
|||||||
{database.TargetTypeLog, true, 0, "", "written to the log", ""},
|
{database.TargetTypeLog, true, 0, "", "written to the log", ""},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Named for type and outcome, so that no two cases share a name:
|
||||||
|
// Go adds "#01" to a repeated one, and a "#" in the temporary
|
||||||
|
// directory's path cuts the path SQLiteDSN gives SQLite short.
|
||||||
for _, tc := range cases {
|
for _, tc := range cases {
|
||||||
t.Run(string(tc.targetType), func(t *testing.T) {
|
t.Run(string(tc.targetType)+" "+tc.outcome, func(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
f := newRecentEventsFixture(t)
|
f := newRecentEventsFixture(t)
|
||||||
|
|||||||
@@ -314,21 +314,15 @@ func assertReplayTask(
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
// assertNoLeakedTarget proves the per-webhook database holds no target
|
// assertNoLeakedTarget proves the per-webhook database holds no
|
||||||
// rows. AutoMigrate creates the table there because Delivery declares
|
// target: an association write would have upserted the whole target,
|
||||||
// the relation, so it is a ROW that signals a leak: an association
|
// plaintext config and all, into the event database. See
|
||||||
// write would have upserted the whole target, plaintext config and
|
|
||||||
// all, into the event database. See
|
|
||||||
// https://git.eeqj.de/sneak/webhooker/issues/206.
|
// https://git.eeqj.de/sneak/webhooker/issues/206.
|
||||||
func assertNoLeakedTarget(t *testing.T, webhookDB *gorm.DB) {
|
func assertNoLeakedTarget(t *testing.T, webhookDB *gorm.DB) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
|
|
||||||
var leaked int64
|
assert.False(
|
||||||
|
t, webhookDB.Migrator().HasTable(&database.Target{}),
|
||||||
require.NoError(t, webhookDB.Unscoped().
|
|
||||||
Model(&database.Target{}).Count(&leaked).Error)
|
|
||||||
assert.Zero(
|
|
||||||
t, leaked,
|
|
||||||
"replay must not write the target into the event database",
|
"replay must not write the target into the event database",
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user