292 lines
13 KiB
Go
292 lines
13 KiB
Go
package sqlite
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"git.zaynet.fr/DoGaMa/DoGaMa-serv/internal/catalog"
|
|
"git.zaynet.fr/DoGaMa/DoGaMa-serv/internal/instance"
|
|
)
|
|
|
|
// Repository persists catalog snapshots and the desired instance registry.
|
|
type Repository struct {
|
|
db *sql.DB
|
|
now func() time.Time
|
|
}
|
|
|
|
func NewRepository(db *sql.DB) *Repository {
|
|
return &Repository{db: db, now: time.Now}
|
|
}
|
|
|
|
func (r *Repository) Sync(ctx context.Context, snapshots []catalog.Snapshot) error {
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("begin catalog sync: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
now := r.now().UTC().Format(time.RFC3339Nano)
|
|
for _, snapshot := range snapshots {
|
|
var digest string
|
|
err := tx.QueryRowContext(ctx, "SELECT digest FROM template_versions WHERE template_id = ? AND version = ?", snapshot.Template.ID, snapshot.Template.Version).Scan(&digest)
|
|
if err == nil && digest != snapshot.Digest {
|
|
return fmt.Errorf("%w: %s@%s", catalog.ErrImmutableSnapshot, snapshot.Template.ID, snapshot.Template.Version)
|
|
}
|
|
if err != nil && !errors.Is(err, sql.ErrNoRows) {
|
|
return fmt.Errorf("check template snapshot: %w", err)
|
|
}
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO templates(id, origin, trust_status, active_version, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(id) DO UPDATE SET active_version=excluded.active_version, updated_at=excluded.updated_at`,
|
|
snapshot.Template.ID, snapshot.Origin, snapshot.Origin, snapshot.Template.Version, now, now)
|
|
if err != nil {
|
|
return fmt.Errorf("upsert catalog template: %w", err)
|
|
}
|
|
if digest == "" {
|
|
_, err = tx.ExecContext(ctx, `INSERT INTO template_versions(template_id, version, schema_version, canonical_yaml, digest, game_id, game_name, description, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, snapshot.Template.ID, snapshot.Template.Version, snapshot.Template.SchemaVersion, snapshot.CanonicalYAML, snapshot.Digest, snapshot.Template.Game.ID, snapshot.Template.Game.Name, snapshot.Template.Game.Description, now)
|
|
if err != nil {
|
|
return fmt.Errorf("insert template snapshot: %w", err)
|
|
}
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit catalog sync: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) List(ctx context.Context) ([]catalog.Summary, error) {
|
|
rows, err := r.db.QueryContext(ctx, `SELECT t.id, t.active_version, v.game_id, v.game_name, v.description, t.trust_status, v.digest
|
|
FROM templates t JOIN template_versions v ON v.template_id=t.id AND v.version=t.active_version ORDER BY v.game_name, t.id`)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list catalog: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var result []catalog.Summary
|
|
for rows.Next() {
|
|
var summary catalog.Summary
|
|
if err := rows.Scan(&summary.ID, &summary.Version, &summary.GameID, &summary.GameName, &summary.Description, &summary.TrustStatus, &summary.Digest); err != nil {
|
|
return nil, fmt.Errorf("scan catalog: %w", err)
|
|
}
|
|
result = append(result, summary)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
func (r *Repository) Get(ctx context.Context, id, version string) (catalog.Snapshot, error) {
|
|
var canonical, digest, origin string
|
|
err := r.db.QueryRowContext(ctx, `SELECT v.canonical_yaml, v.digest, t.origin FROM template_versions v JOIN templates t ON t.id=v.template_id
|
|
WHERE v.template_id=? AND v.version=?`, id, version).Scan(&canonical, &digest, &origin)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return catalog.Snapshot{}, catalog.ErrTemplateNotFound
|
|
}
|
|
if err != nil {
|
|
return catalog.Snapshot{}, fmt.Errorf("load template snapshot: %w", err)
|
|
}
|
|
var template catalog.Template
|
|
if err := json.Unmarshal([]byte(canonical), &template); err != nil {
|
|
return catalog.Snapshot{}, fmt.Errorf("decode stored template snapshot: %w", err)
|
|
}
|
|
return catalog.Snapshot{Template: template, CanonicalYAML: canonical, Digest: digest, Origin: origin}, nil
|
|
}
|
|
|
|
func (r *Repository) CreateDraft(ctx context.Context, draft instance.Draft) error {
|
|
if draft.ID == "" {
|
|
return errors.New("draft instance ID is required")
|
|
}
|
|
now := r.now().UTC().Format(time.RFC3339Nano)
|
|
previewJSON, err := json.Marshal(draft.Preview)
|
|
if err != nil {
|
|
return fmt.Errorf("encode draft preview: %w", err)
|
|
}
|
|
_, err = r.db.ExecContext(ctx, `INSERT INTO instances(id, slug, display_name, template_id, template_version, template_digest, revision, lifecycle_state, preview_json, plan_digest, custom_labels_json, docker_user_mode, docker_uid, docker_gid, image_tag_mode, image_tag, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, 1, 'draft', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, draft.ID, draft.Preview.Slug, draft.Preview.DisplayName, draft.Preview.Template.ID, draft.Preview.Template.Version, draft.Preview.Template.Digest, string(previewJSON), draft.Preview.PlanDigest, string(mustJSON(draft.Preview.CustomLabels)), draft.Preview.DockerUser.Mode, draft.Preview.DockerUser.UID, draft.Preview.DockerUser.GID, draft.Preview.ImageTag.Mode, draft.Preview.ImageTag.Tag, now, now)
|
|
if err != nil {
|
|
return fmt.Errorf("create draft instance: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) GetInstance(ctx context.Context, id string) (instance.StoredInstance, error) {
|
|
return scanInstance(r.db.QueryRowContext(ctx, `SELECT id, preview_json, lifecycle_state, observed_state, COALESCE(container_id, ''), plan_digest, desired_running, container_config_pending FROM instances WHERE id=? AND deleted_at IS NULL`, id))
|
|
}
|
|
|
|
func (r *Repository) ListLifecycleInstances(ctx context.Context) ([]instance.StoredInstance, error) {
|
|
rows, err := r.db.QueryContext(ctx, `SELECT id, preview_json, lifecycle_state, observed_state, COALESCE(container_id, ''), plan_digest, desired_running, container_config_pending FROM instances WHERE deleted_at IS NULL AND lifecycle_state != 'draft' ORDER BY id`)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list lifecycle instances: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var result []instance.StoredInstance
|
|
for rows.Next() {
|
|
value, err := scanInstance(rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
result = append(result, value)
|
|
}
|
|
return result, rows.Err()
|
|
}
|
|
|
|
type rowScanner interface{ Scan(...any) error }
|
|
|
|
func scanInstance(row rowScanner) (instance.StoredInstance, error) {
|
|
var value instance.StoredInstance
|
|
var previewJSON string
|
|
var desired int
|
|
var pending int
|
|
err := row.Scan(&value.ID, &previewJSON, &value.LifecycleState, &value.ObservedState, &value.ContainerID, &value.PlanDigest, &desired, &pending)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return instance.StoredInstance{}, instance.ErrInstanceNotFound
|
|
}
|
|
if err != nil {
|
|
return instance.StoredInstance{}, fmt.Errorf("scan instance: %w", err)
|
|
}
|
|
if err := json.Unmarshal([]byte(previewJSON), &value.Preview); err != nil {
|
|
return instance.StoredInstance{}, fmt.Errorf("decode instance preview: %w", err)
|
|
}
|
|
value.DesiredRunning = desired != 0
|
|
value.ContainerConfigPending = pending != 0
|
|
return value, nil
|
|
}
|
|
|
|
func (r *Repository) BeginOperation(ctx context.Context, operationID, instanceID, kind, lifecycleState string) (instance.StoredInstance, error) {
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return instance.StoredInstance{}, fmt.Errorf("begin instance operation: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
current, err := scanInstance(tx.QueryRowContext(ctx, `SELECT id, preview_json, lifecycle_state, observed_state, COALESCE(container_id, ''), plan_digest, desired_running, container_config_pending FROM instances WHERE id=? AND deleted_at IS NULL`, instanceID))
|
|
if err != nil {
|
|
return instance.StoredInstance{}, err
|
|
}
|
|
var active int
|
|
err = tx.QueryRowContext(ctx, `SELECT 1 FROM instance_operations WHERE instance_id=? AND state='running' LIMIT 1`, instanceID).Scan(&active)
|
|
if err == nil {
|
|
return instance.StoredInstance{}, instance.ErrOperationConflict
|
|
}
|
|
if !errors.Is(err, sql.ErrNoRows) {
|
|
return instance.StoredInstance{}, fmt.Errorf("check active operation: %w", err)
|
|
}
|
|
now := r.now().UTC().Format(time.RFC3339Nano)
|
|
if _, err := tx.ExecContext(ctx, `INSERT INTO instance_operations(id, instance_id, kind, state, phase, created_at, updated_at) VALUES (?, ?, ?, 'running', 'dispatch', ?, ?)`, operationID, instanceID, kind, now, now); err != nil {
|
|
return instance.StoredInstance{}, fmt.Errorf("insert instance operation: %w", err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `UPDATE instances SET lifecycle_state=?, last_error_code=NULL, updated_at=? WHERE id=?`, lifecycleState, now, instanceID); err != nil {
|
|
return instance.StoredInstance{}, fmt.Errorf("mark instance operation: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return instance.StoredInstance{}, fmt.Errorf("commit instance operation: %w", err)
|
|
}
|
|
current.LifecycleState = lifecycleState
|
|
return current, nil
|
|
}
|
|
|
|
func (r *Repository) FinishOperation(ctx context.Context, operationID, lifecycleState, observedState, containerID, planDigest string, desiredRunning bool, errorCode string) error {
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("begin operation completion: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
now := r.now().UTC().Format(time.RFC3339Nano)
|
|
desired := 0
|
|
if desiredRunning {
|
|
desired = 1
|
|
}
|
|
result, err := tx.ExecContext(ctx, `UPDATE instance_operations SET state='succeeded', phase='complete', error_code=NULL, updated_at=?, completed_at=? WHERE id=? AND state='running'`, now, now, operationID)
|
|
if err != nil {
|
|
return fmt.Errorf("complete instance operation: %w", err)
|
|
}
|
|
changed, _ := result.RowsAffected()
|
|
if changed != 1 {
|
|
return instance.ErrOperationConflict
|
|
}
|
|
var container any
|
|
if containerID != "" {
|
|
container = containerID
|
|
}
|
|
_, err = tx.ExecContext(ctx, `UPDATE instances SET lifecycle_state=?, observed_state=?, container_id=?, plan_digest=?, desired_running=?, last_error_code=?, updated_at=? WHERE id=(SELECT instance_id FROM instance_operations WHERE id=?)`, lifecycleState, observedState, container, planDigest, desired, nullable(errorCode), now, operationID)
|
|
if err != nil {
|
|
return fmt.Errorf("update completed instance: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit operation completion: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) FailOperation(ctx context.Context, operationID, lifecycleState, errorCode string) error {
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("begin operation failure: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
now := r.now().UTC().Format(time.RFC3339Nano)
|
|
result, err := tx.ExecContext(ctx, `UPDATE instance_operations SET state='failed', phase='failed', error_code=?, updated_at=?, completed_at=? WHERE id=? AND state='running'`, errorCode, now, now, operationID)
|
|
if err != nil {
|
|
return fmt.Errorf("fail instance operation: %w", err)
|
|
}
|
|
changed, _ := result.RowsAffected()
|
|
if changed != 1 {
|
|
return instance.ErrOperationConflict
|
|
}
|
|
_, err = tx.ExecContext(ctx, `UPDATE instances SET lifecycle_state=?, observed_state='unknown', last_error_code=?, updated_at=? WHERE id=(SELECT instance_id FROM instance_operations WHERE id=?)`, lifecycleState, errorCode, now, operationID)
|
|
if err != nil {
|
|
return fmt.Errorf("update failed instance: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit operation failure: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) UpdateObservation(ctx context.Context, instanceID, lifecycleState, observedState, containerID string, desiredRunning bool, errorCode string) error {
|
|
desired := 0
|
|
if desiredRunning {
|
|
desired = 1
|
|
}
|
|
var container any
|
|
if containerID != "" {
|
|
container = containerID
|
|
}
|
|
result, err := r.db.ExecContext(ctx, `UPDATE instances SET lifecycle_state=?, observed_state=?, container_id=?, desired_running=?, last_error_code=?, updated_at=? WHERE id=? AND deleted_at IS NULL`, lifecycleState, observedState, container, desired, nullable(errorCode), r.now().UTC().Format(time.RFC3339Nano), instanceID)
|
|
if err != nil {
|
|
return fmt.Errorf("update instance observation: %w", err)
|
|
}
|
|
changed, _ := result.RowsAffected()
|
|
if changed != 1 {
|
|
return instance.ErrInstanceNotFound
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *Repository) RecoverInterruptedOperations(ctx context.Context) error {
|
|
tx, err := r.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("begin interrupted-operation recovery: %w", err)
|
|
}
|
|
defer func() { _ = tx.Rollback() }()
|
|
now := r.now().UTC().Format(time.RFC3339Nano)
|
|
if _, err := tx.ExecContext(ctx, `UPDATE instances SET lifecycle_state='intervention_required', last_error_code='operation_interrupted', updated_at=? WHERE id IN (SELECT instance_id FROM instance_operations WHERE state='running')`, now); err != nil {
|
|
return fmt.Errorf("mark interrupted instances: %w", err)
|
|
}
|
|
if _, err := tx.ExecContext(ctx, `UPDATE instance_operations SET state='intervention_required', phase='interrupted', error_code='operation_interrupted', updated_at=?, completed_at=? WHERE state='running'`, now, now); err != nil {
|
|
return fmt.Errorf("mark interrupted operations: %w", err)
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return fmt.Errorf("commit interrupted-operation recovery: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func nullable(value string) any {
|
|
if value == "" {
|
|
return nil
|
|
}
|
|
return value
|
|
}
|