agent-radar
2ed3415fb7
Publish Claude Code Binaries / Stage and publish binaries (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-arm64 (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build linux-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build windows-amd64 (push) Canceled after 0s
Sync Tutti Agent Skills / sync (push) Canceled after 0s
Windows Agent Adapters / Test Windows agent adapters (push) Canceled after 0s
Windows Daemon Adapters / Test Windows daemon adapters (push) Canceled after 0s
Publish Tutti App Runtime / Publish runtime catalog (push) Canceled after 0s
619 lines
23 KiB
Go
619 lines
23 KiB
Go
package storesqlite
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"path/filepath"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
market "github.com/tutti-os/tutti/packages/connector/daemon/core"
|
|
)
|
|
|
|
func TestStoreCleanupLifecycleHonorsCutoffsBatchesAndProtectedRows(t *testing.T) {
|
|
ctx := context.Background()
|
|
store, err := Open(ctx, filepath.Join(t.TempDir(), "tuttid.db"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer store.Close()
|
|
|
|
cutoff := time.Date(2026, 8, 6, 12, 0, 0, 0, time.UTC)
|
|
operations := []market.Operation{
|
|
lifecycleTestOperation("completed-before", "connector-completed", market.OperationStateCompleted, cutoff.Add(-time.Millisecond)),
|
|
lifecycleTestOperation("failed-at", "connector-failed", market.OperationStateFailed, cutoff),
|
|
lifecycleTestOperation("completed-after", "connector-newer", market.OperationStateCompleted, cutoff.Add(time.Millisecond)),
|
|
lifecycleTestOperation("accepted-old", "connector-accepted", market.OperationStateAccepted, cutoff.Add(-time.Hour)),
|
|
lifecycleTestOperation("running-old", "connector-running", market.OperationStateRunning, cutoff.Add(-time.Hour)),
|
|
}
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
for index, operation := range operations {
|
|
if err := tx.SaveOperation(operation); err != nil {
|
|
return err
|
|
}
|
|
if err := tx.EnqueueConnectorMarketChanged(market.ChangedEvent{OperationID: operation.OperationID, Revision: uint64(index + 1)}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for sequence, publishedAt := range map[int64]time.Time{
|
|
1: cutoff.Add(-time.Millisecond),
|
|
2: cutoff,
|
|
3: cutoff.Add(time.Millisecond),
|
|
5: cutoff.Add(2 * time.Millisecond),
|
|
} {
|
|
if err := store.MarkChangedEventPublished(ctx, sequence, publishedAt); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
request := market.LifecycleCleanupRequest{
|
|
TerminalOperationsUpdatedThrough: cutoff,
|
|
PublishedEventsPublishedThrough: cutoff,
|
|
BatchSize: 1,
|
|
}
|
|
first, err := store.CleanupLifecycle(ctx, request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
second, err := store.CleanupLifecycle(ctx, request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
third, err := store.CleanupLifecycle(ctx, request)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if first.TerminalOperationsDeleted != 1 || first.PublishedEventsDeleted != 1 ||
|
|
second.TerminalOperationsDeleted != 1 || second.PublishedEventsDeleted != 1 ||
|
|
third.TerminalOperationsDeleted != 0 || third.PublishedEventsDeleted != 0 {
|
|
t.Fatalf("cleanup results = %#v, %#v, %#v", first, second, third)
|
|
}
|
|
for _, operationID := range []string{"completed-before", "failed-at"} {
|
|
if _, err := store.Operation(ctx, operationID); !errors.Is(err, market.ErrNotFound) {
|
|
t.Fatalf("operation %q error = %v, want not found", operationID, err)
|
|
}
|
|
}
|
|
for _, operationID := range []string{"completed-after", "accepted-old", "running-old"} {
|
|
if _, err := store.Operation(ctx, operationID); err != nil {
|
|
t.Fatalf("protected operation %q error = %v", operationID, err)
|
|
}
|
|
}
|
|
reusedRequest := lifecycleTestOperation("accepted-reused-request", "connector-reused-request", market.OperationStateAccepted, cutoff.Add(time.Minute))
|
|
reusedRequest.ClientRequestID = operations[0].ClientRequestID
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
return tx.SaveOperation(reusedRequest)
|
|
}); err != nil {
|
|
t.Fatalf("reuse expired client request ID: %v", err)
|
|
}
|
|
pending, err := store.PendingChangedEvents(ctx, 10)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(pending) != 1 || pending[0].Sequence != 4 {
|
|
t.Fatalf("pending events = %#v, want only sequence 4", pending)
|
|
}
|
|
var publishedRows int
|
|
if err := store.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM connector_market_outbox WHERE published_at_unix_ms IS NOT NULL`).Scan(&publishedRows); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if publishedRows != 2 {
|
|
t.Fatalf("published outbox rows = %d, want 2 protected newer rows", publishedRows)
|
|
}
|
|
if _, err := store.CleanupLifecycle(ctx, market.LifecycleCleanupRequest{
|
|
TerminalOperationsUpdatedThrough: cutoff,
|
|
PublishedEventsPublishedThrough: cutoff,
|
|
BatchSize: maxLifecycleCleanupBatchSize + 1,
|
|
}); err == nil {
|
|
t.Fatal("oversized lifecycle cleanup batch was accepted")
|
|
}
|
|
}
|
|
|
|
func TestStoreCleanupLifecycleIsSafeAcrossConnections(t *testing.T) {
|
|
ctx := context.Background()
|
|
databasePath := filepath.Join(t.TempDir(), "tuttid.db")
|
|
first, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer first.Close()
|
|
second, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer second.Close()
|
|
cutoff := time.Date(2026, 8, 6, 12, 0, 0, 0, time.UTC)
|
|
if err := first.Transaction(ctx, func(tx market.Transaction) error {
|
|
for index := range 20 {
|
|
operation := lifecycleTestOperation(
|
|
fmt.Sprintf("terminal-%02d", index),
|
|
fmt.Sprintf("connector-%02d", index),
|
|
market.OperationStateCompleted,
|
|
cutoff.Add(-time.Hour),
|
|
)
|
|
if err := tx.SaveOperation(operation); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
request := market.LifecycleCleanupRequest{
|
|
TerminalOperationsUpdatedThrough: cutoff,
|
|
PublishedEventsPublishedThrough: cutoff,
|
|
BatchSize: 7,
|
|
}
|
|
start := make(chan struct{})
|
|
results := make(chan market.LifecycleCleanupResult, 2)
|
|
errorsFound := make(chan error, 2)
|
|
var wait sync.WaitGroup
|
|
for _, candidate := range []*Store{first, second} {
|
|
wait.Add(1)
|
|
go func(store *Store) {
|
|
defer wait.Done()
|
|
<-start
|
|
result, cleanupErr := store.CleanupLifecycle(ctx, request)
|
|
if cleanupErr != nil {
|
|
errorsFound <- cleanupErr
|
|
return
|
|
}
|
|
results <- result
|
|
}(candidate)
|
|
}
|
|
close(start)
|
|
wait.Wait()
|
|
close(results)
|
|
close(errorsFound)
|
|
for cleanupErr := range errorsFound {
|
|
t.Fatal(cleanupErr)
|
|
}
|
|
var deleted int64
|
|
for result := range results {
|
|
if result.TerminalOperationsDeleted > int64(request.BatchSize) {
|
|
t.Fatalf("one cleanup deleted %d operations, batch size %d", result.TerminalOperationsDeleted, request.BatchSize)
|
|
}
|
|
deleted += result.TerminalOperationsDeleted
|
|
}
|
|
if deleted != 14 {
|
|
t.Fatalf("concurrent cleanup deleted %d operations, want 14", deleted)
|
|
}
|
|
var remaining int
|
|
if err := first.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM connector_market_operations`).Scan(&remaining); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if remaining != 6 {
|
|
t.Fatalf("remaining operations = %d, want 6", remaining)
|
|
}
|
|
}
|
|
|
|
func TestStoreCleanupPreservesRecoverableWorkAcrossReopen(t *testing.T) {
|
|
ctx := context.Background()
|
|
databasePath := filepath.Join(t.TempDir(), "tuttid.db")
|
|
store, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
operation := lifecycleTestOperation(
|
|
"accepted-restart",
|
|
"connector-restart",
|
|
market.OperationStateAccepted,
|
|
time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC),
|
|
)
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
if err := tx.SaveOperation(operation); err != nil {
|
|
return err
|
|
}
|
|
return tx.EnqueueConnectorMarketChanged(market.ChangedEvent{
|
|
ConnectorKey: operation.ConnectorKey,
|
|
OperationID: operation.OperationID,
|
|
Revision: 1,
|
|
})
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
cutoff := time.Date(2026, 8, 6, 12, 0, 0, 0, time.UTC)
|
|
if _, err := store.CleanupLifecycle(ctx, market.LifecycleCleanupRequest{
|
|
TerminalOperationsUpdatedThrough: cutoff,
|
|
PublishedEventsPublishedThrough: cutoff,
|
|
BatchSize: 10,
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := store.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
reopened, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer reopened.Close()
|
|
recoverable, err := reopened.RecoverableOperations(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(recoverable) != 1 || recoverable[0].OperationID != operation.OperationID {
|
|
t.Fatalf("recoverable operations = %#v", recoverable)
|
|
}
|
|
pending, err := reopened.PendingChangedEvents(ctx, 10)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(pending) != 1 || pending[0].Event.OperationID != "" || pending[0].Event.ConnectorKey != operation.ConnectorKey {
|
|
t.Fatalf("pending events = %#v", pending)
|
|
}
|
|
}
|
|
|
|
func TestStoreInstalledReleaseEvidenceSurvivesTerminalCleanupAndReopen(t *testing.T) {
|
|
ctx := context.Background()
|
|
databasePath := filepath.Join(t.TempDir(), "tuttid.db")
|
|
store, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
release := testConnector().Release
|
|
connector := testConnector()
|
|
connector.Installation = market.Installation{
|
|
State: market.InstallationStateInstalled,
|
|
InstalledVersion: release.Version,
|
|
InstalledReleaseID: release.ReleaseID,
|
|
InstalledReleaseDigest: release.ReleaseDigest,
|
|
}
|
|
completedAt := time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC)
|
|
operation := lifecycleTestOperation("install-evidence", connector.Key, market.OperationStateCompleted, completedAt)
|
|
operation.Kind = market.OperationKindInstall
|
|
operation.Target = &market.OperationTarget{
|
|
ConnectorKey: connector.Key, Version: release.Version, ReleaseID: release.ReleaseID,
|
|
ReleaseDigest: release.ReleaseDigest, ArtifactSHA256: release.Artifact.SHA256, Release: &release,
|
|
}
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
connector.Revision = tx.AdvanceRevision()
|
|
if err := tx.SaveConnector(connector); err != nil {
|
|
return err
|
|
}
|
|
return tx.SaveOperation(operation)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
connector.Release.Version = "2.0.0"
|
|
connector.Release.ReleaseID = "43"
|
|
connector.Release.ReleaseDigest = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
|
|
connector.Release.ManifestDigest = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
connector.Revision = tx.AdvanceRevision()
|
|
return tx.SaveConnector(connector)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
result, err := store.CleanupLifecycle(ctx, market.LifecycleCleanupRequest{
|
|
TerminalOperationsUpdatedThrough: completedAt,
|
|
PublishedEventsPublishedThrough: completedAt,
|
|
BatchSize: 10,
|
|
})
|
|
if err != nil || result.TerminalOperationsDeleted != 1 {
|
|
t.Fatalf("cleanup result = %#v, error = %v", result, err)
|
|
}
|
|
if _, err := store.Operation(ctx, operation.OperationID); !errors.Is(err, market.ErrNotFound) {
|
|
t.Fatalf("cleaned operation error = %v", err)
|
|
}
|
|
if err := store.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
reopened, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer reopened.Close()
|
|
got, err := reopened.InstalledRelease(ctx, connector.Key, release.ReleaseDigest)
|
|
if err != nil || got.ReleaseDigest != release.ReleaseDigest || got.ManifestDigest != release.ManifestDigest {
|
|
t.Fatalf("installed release = %#v, error = %v", got, err)
|
|
}
|
|
}
|
|
|
|
func TestStoreCompletedLocalUninstallDeletesReleaseEvidenceButKeepsAuthorizationProjection(t *testing.T) {
|
|
ctx := context.Background()
|
|
store, err := Open(ctx, filepath.Join(t.TempDir(), "tuttid.db"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer store.Close()
|
|
release := testConnector().Release
|
|
connector := testConnector()
|
|
connector.Installation = market.Installation{State: market.InstallationStateInstalled,
|
|
InstalledVersion: release.Version, InstalledReleaseID: release.ReleaseID, InstalledReleaseDigest: release.ReleaseDigest}
|
|
installedAt := time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC)
|
|
install := lifecycleTestOperation("install-before-local-uninstall", connector.Key, market.OperationStateCompleted, installedAt)
|
|
install.Kind = market.OperationKindInstall
|
|
install.Target = &market.OperationTarget{ConnectorKey: connector.Key, Version: release.Version, ReleaseID: release.ReleaseID,
|
|
ReleaseDigest: release.ReleaseDigest, ArtifactSHA256: release.Artifact.SHA256, Release: &release}
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
connector.Revision = tx.AdvanceRevision()
|
|
if err := tx.SaveConnector(connector); err != nil {
|
|
return err
|
|
}
|
|
return tx.SaveOperation(install)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
projection := market.AuthorizationProjection{AccountID: "account-1", ConnectorKey: connector.Key,
|
|
ConnectionID: "connection-1", State: market.AuthorizationStateConnected, ServerRevision: 8,
|
|
ServerSynchronized: true, UpdatedAt: installedAt}
|
|
if err := store.SaveAuthorizationProjection(ctx, projection); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
uninstall := lifecycleTestOperation("local-uninstall", connector.Key, market.OperationStateCompleted, installedAt.Add(time.Minute))
|
|
uninstall.Kind = market.OperationKindUninstall
|
|
uninstall.Target = &market.OperationTarget{ConnectorKey: connector.Key, Version: release.Version,
|
|
ReleaseID: release.ReleaseID, ReleaseDigest: release.ReleaseDigest}
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
connector.Installation = market.Installation{State: market.InstallationStateNotInstalled}
|
|
connector.Revision = tx.AdvanceRevision()
|
|
if err := tx.SaveConnector(connector); err != nil {
|
|
return err
|
|
}
|
|
return tx.SaveOperation(uninstall)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := store.InstalledRelease(ctx, connector.Key, release.ReleaseDigest); !errors.Is(err, market.ErrNotFound) {
|
|
t.Fatalf("installed release error = %v, want not found", err)
|
|
}
|
|
storedProjection, err := store.AuthorizationProjection(ctx, projection.AccountID, projection.ConnectorKey)
|
|
if err != nil || storedProjection != projection {
|
|
t.Fatalf("authorization projection = %#v, error = %v", storedProjection, err)
|
|
}
|
|
}
|
|
|
|
func TestStoreRetainsCurrentAndPreparedCandidateReleaseEvidence(t *testing.T) {
|
|
ctx := context.Background()
|
|
store, err := Open(ctx, filepath.Join(t.TempDir(), "tuttid.db"))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer store.Close()
|
|
|
|
connector := testConnector()
|
|
current := connector.Release
|
|
connector.Installation = market.Installation{
|
|
State: market.InstallationStateUpdating, InstalledVersion: current.Version,
|
|
InstalledReleaseID: current.ReleaseID, InstalledReleaseDigest: current.ReleaseDigest,
|
|
}
|
|
candidate := current
|
|
candidate.Version = "2.0.0"
|
|
candidate.ReleaseID = connector.Key + "@2.0.0"
|
|
candidate.ReleaseDigest = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
|
|
candidate.ManifestDigest = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
|
|
installedAt := time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC)
|
|
currentOperation := lifecycleTestOperation("current-release", connector.Key, market.OperationStateCompleted, installedAt)
|
|
currentOperation.Kind = market.OperationKindInstall
|
|
currentOperation.Target = &market.OperationTarget{
|
|
ConnectorKey: connector.Key, Version: current.Version, ReleaseID: current.ReleaseID,
|
|
ReleaseDigest: current.ReleaseDigest, ArtifactSHA256: current.Artifact.SHA256, Release: ¤t,
|
|
}
|
|
candidateOperation := lifecycleTestOperation("candidate-release", connector.Key, market.OperationStateRunning, installedAt.Add(time.Minute))
|
|
candidateOperation.Kind = market.OperationKindInstall
|
|
candidateOperation.Stage = market.OperationStageRuntimePending
|
|
candidateOperation.Target = &market.OperationTarget{
|
|
ConnectorKey: connector.Key, Version: candidate.Version, ReleaseID: candidate.ReleaseID,
|
|
ReleaseDigest: candidate.ReleaseDigest, ArtifactSHA256: candidate.Artifact.SHA256, Release: &candidate,
|
|
}
|
|
candidateOperation.Execution.ReleaseInstallation = &market.ReleaseInstallationReceipt{
|
|
OperationID: candidateOperation.OperationID, ConnectorKey: connector.Key,
|
|
Version: candidate.Version, ReleaseID: candidate.ReleaseID, ReleaseDigest: candidate.ReleaseDigest,
|
|
}
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
connector.Revision = tx.AdvanceRevision()
|
|
if err := tx.SaveConnector(connector); err != nil {
|
|
return err
|
|
}
|
|
if err := tx.SaveOperation(currentOperation); err != nil {
|
|
return err
|
|
}
|
|
return tx.SaveOperation(candidateOperation)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var currentDigest string
|
|
if err := store.db.QueryRowContext(ctx, `SELECT release_digest FROM connector_market_installed_releases
|
|
WHERE connector_key = ?`, connector.Key).Scan(¤tDigest); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if currentDigest != current.ReleaseDigest {
|
|
t.Fatalf("current release pointer = %q, want %q", currentDigest, current.ReleaseDigest)
|
|
}
|
|
|
|
for name, expected := range map[string]market.Release{"current": current, "candidate": candidate} {
|
|
got, err := store.InstalledRelease(ctx, connector.Key, expected.ReleaseDigest)
|
|
if err != nil || got.ReleaseDigest != expected.ReleaseDigest || got.ManifestDigest != expected.ManifestDigest {
|
|
t.Fatalf("%s release = %#v, error = %v", name, got, err)
|
|
}
|
|
}
|
|
|
|
candidateOperation.State = market.OperationStateCompleted
|
|
candidateOperation.Stage = market.OperationStageCompleted
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
return tx.SaveOperation(candidateOperation)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := store.db.QueryRowContext(ctx, `SELECT release_digest FROM connector_market_installed_releases
|
|
WHERE connector_key = ?`, connector.Key).Scan(¤tDigest); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if currentDigest != candidate.ReleaseDigest {
|
|
t.Fatalf("promoted release pointer = %q, want %q", currentDigest, candidate.ReleaseDigest)
|
|
}
|
|
}
|
|
|
|
func TestStoreMigrationBackfillsLifecycleTimestampAndInstalledReleaseEvidence(t *testing.T) {
|
|
ctx := context.Background()
|
|
databasePath := filepath.Join(t.TempDir(), "tuttid.db")
|
|
legacy, err := sql.Open("sqlite", databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, statement := range []string{
|
|
`CREATE TABLE connector_market_connectors (connector_key TEXT PRIMARY KEY, connector_json TEXT NOT NULL)`,
|
|
`CREATE TABLE connector_market_installed_releases (
|
|
connector_key TEXT PRIMARY KEY,
|
|
release_digest TEXT NOT NULL,
|
|
release_json TEXT NOT NULL
|
|
)`,
|
|
`CREATE TABLE connector_market_operations (
|
|
operation_id TEXT PRIMARY KEY,
|
|
client_request_id TEXT NOT NULL UNIQUE,
|
|
connector_key TEXT NOT NULL,
|
|
kind TEXT NOT NULL,
|
|
state TEXT NOT NULL,
|
|
lease_owner TEXT NOT NULL,
|
|
lease_token INTEGER NOT NULL DEFAULT 0,
|
|
lease_expires_at_unix_ms INTEGER,
|
|
operation_json TEXT NOT NULL
|
|
)`,
|
|
} {
|
|
if _, err := legacy.ExecContext(ctx, statement); err != nil {
|
|
_ = legacy.Close()
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
installedRelease := testConnector().Release
|
|
connector := testConnector()
|
|
connector.Installation = market.Installation{
|
|
State: market.InstallationStateInstalled,
|
|
InstalledVersion: installedRelease.Version,
|
|
InstalledReleaseID: installedRelease.ReleaseID,
|
|
InstalledReleaseDigest: installedRelease.ReleaseDigest,
|
|
}
|
|
connector.Release.Version = "2.0.0"
|
|
connector.Release.ReleaseDigest = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
|
|
operation := lifecycleTestOperation(
|
|
"legacy-install",
|
|
connector.Key,
|
|
market.OperationStateCompleted,
|
|
time.Date(2026, 8, 5, 12, 0, 0, 0, time.UTC),
|
|
)
|
|
operation.Kind = market.OperationKindInstall
|
|
operation.Target = &market.OperationTarget{
|
|
ConnectorKey: connector.Key, Version: installedRelease.Version, ReleaseID: installedRelease.ReleaseID,
|
|
ReleaseDigest: installedRelease.ReleaseDigest, ArtifactSHA256: installedRelease.Artifact.SHA256, Release: &installedRelease,
|
|
}
|
|
connectorPayload, err := json.Marshal(connector)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
operationPayload, err := json.Marshal(operation)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := legacy.ExecContext(ctx, `INSERT INTO connector_market_connectors (connector_key, connector_json) VALUES (?, ?)`,
|
|
connector.Key, string(connectorPayload)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
installedReleasePayload, err := json.Marshal(installedRelease)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := legacy.ExecContext(ctx, `INSERT INTO connector_market_installed_releases
|
|
(connector_key, release_digest, release_json) VALUES (?, ?, ?)`, connector.Key, installedRelease.ReleaseDigest,
|
|
string(installedReleasePayload)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := legacy.ExecContext(ctx, `INSERT INTO connector_market_operations (
|
|
operation_id, client_request_id, connector_key, kind, state, lease_owner, lease_token, lease_expires_at_unix_ms, operation_json
|
|
) VALUES (?, ?, ?, ?, ?, '', 0, NULL, ?)`, operation.OperationID, operation.ClientRequestID, operation.ConnectorKey,
|
|
operation.Kind, operation.State, string(operationPayload)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := legacy.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
store, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer store.Close()
|
|
var updatedAtMS int64
|
|
if err := store.db.QueryRowContext(ctx, `SELECT updated_at_unix_ms FROM connector_market_operations WHERE operation_id = ?`, operation.OperationID).Scan(&updatedAtMS); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if updatedAtMS != operation.UpdatedAt.UnixMilli() {
|
|
t.Fatalf("migrated updated timestamp = %d, want %d", updatedAtMS, operation.UpdatedAt.UnixMilli())
|
|
}
|
|
migratedConnector, err := store.Connector(ctx, connector.Key)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if migratedConnector.Installation.InstalledAtUnixMS != operation.UpdatedAt.UnixMilli() {
|
|
t.Fatalf("migrated install timestamp = %d, want %d", migratedConnector.Installation.InstalledAtUnixMS, operation.UpdatedAt.UnixMilli())
|
|
}
|
|
release, err := store.InstalledRelease(ctx, connector.Key, installedRelease.ReleaseDigest)
|
|
if err != nil || release.ReleaseDigest != installedRelease.ReleaseDigest {
|
|
t.Fatalf("migrated installed release = %#v, error = %v", release, err)
|
|
}
|
|
var historicalCount int
|
|
if err := store.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM connector_market_release_installations
|
|
WHERE connector_key = ? AND release_digest = ?`, connector.Key, installedRelease.ReleaseDigest).Scan(&historicalCount); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if historicalCount != 1 {
|
|
t.Fatalf("migrated release installation evidence count = %d, want 1", historicalCount)
|
|
}
|
|
}
|
|
|
|
func TestStoreReopenToleratesMissingLegacyInstalledReleaseEvidence(t *testing.T) {
|
|
ctx := context.Background()
|
|
databasePath := filepath.Join(t.TempDir(), "tuttid.db")
|
|
store, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
connector := testConnector()
|
|
installedRelease := connector.Release
|
|
connector.Installation = market.Installation{
|
|
State: market.InstallationStateInstalled, InstalledVersion: installedRelease.Version,
|
|
InstalledReleaseID: installedRelease.ReleaseID, InstalledReleaseDigest: installedRelease.ReleaseDigest,
|
|
}
|
|
connector.Release.Version = "2.0.0"
|
|
connector.Release.ReleaseID = connector.Key + "@2.0.0"
|
|
connector.Release.ReleaseDigest = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"
|
|
connector.Release.ManifestDigest = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee"
|
|
if err := store.Transaction(ctx, func(tx market.Transaction) error {
|
|
connector.Revision = tx.AdvanceRevision()
|
|
return tx.SaveConnector(connector)
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := store.Close(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
reopened, err := Open(ctx, databasePath)
|
|
if err != nil {
|
|
t.Fatalf("reopen with missing legacy evidence: %v", err)
|
|
}
|
|
defer reopened.Close()
|
|
if _, err := reopened.Connector(ctx, connector.Key); err != nil {
|
|
t.Fatalf("load connector after compatible reopen: %v", err)
|
|
}
|
|
if _, err := reopened.InstalledRelease(ctx, connector.Key, installedRelease.ReleaseDigest); !errors.Is(err, market.ErrNotFound) {
|
|
t.Fatalf("missing release evidence error = %v, want not found", err)
|
|
}
|
|
}
|
|
|
|
func lifecycleTestOperation(operationID, connectorKey string, state market.OperationState, updatedAt time.Time) market.Operation {
|
|
return market.Operation{
|
|
OperationID: operationID, ClientRequestID: "request-" + operationID, ConnectorKey: connectorKey,
|
|
Kind: market.OperationKindRefreshCatalog, State: state, Stage: market.OperationStageCompleted,
|
|
CreatedAt: updatedAt.Add(-time.Minute), UpdatedAt: updatedAt,
|
|
}
|
|
}
|