tiny-forge/internal/workload/plugin/source/static/state.go

package static

import (
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"sync"

	"github.com/alexei/tinyforge/internal/store"
	"github.com/alexei/tinyforge/internal/workload/plugin"
)

// runtimeState is the per-workload state that the legacy static_sites
// table used to track on its own row (last_commit_sha, last_sync_at,
// status/error). With the cutover off the synthetic-row adapter these
// fields live inside the container row's extra_json blob keyed by the
// deterministic row ID `<workloadID>:site`.
//
// Unknown keys in extra_json are preserved across read+write so future
// writers (e.g. per-face route maps) can extend the blob without
// forcing this struct to grow. Decoding into a typed wrapper on its
// own would silently drop them; saveState round-trips through a
// generic map first, then merges the typed fields.
type runtimeState struct {
	LastCommitSHA string `json:"last_commit_sha,omitempty"`
	LastSyncAt    string `json:"last_sync_at,omitempty"`
	LastError     string `json:"last_error,omitempty"`
	// Status mirrors the legacy static_sites.status column ("syncing",
	// "deployed", "failed", "stopped"). Kept in extra_json so callers
	// can still answer "is this site healthy?" without a Docker probe.
	Status string `json:"status,omitempty"`
}

// runtimeStateKeys lists every JSON field name owned by runtimeState.
// saveState strips these from the generic map before re-emitting so
// the typed values don't double-write under both their JSON tag and
// any subsequent extension's tag — and so that clearing a field
// (LastError → "") actually removes the key instead of being shadowed
// by a stale carry-over.
var runtimeStateKeys = []string{"last_commit_sha", "last_sync_at", "last_error", "status"}

// containerRowID is the deterministic ID for the single container row
// owned by a static workload. Stable across redeploys so saveState can
// upsert in place.
func containerRowID(w plugin.Workload) string {
	return w.ID + ":site"
}

// loadState returns the persisted runtime state plus the underlying
// container row. Both values are zero on first deploy (no row yet);
// callers must tolerate a nil container without treating it as an
// error.
func loadState(deps plugin.Deps, w plugin.Workload) (runtimeState, *store.Container, error) {
	row, err := deps.Store.GetContainerByID(containerRowID(w))
	if err != nil {
		if errors.Is(err, store.ErrNotFound) {
			return runtimeState{}, nil, nil
		}
		return runtimeState{}, nil, fmt.Errorf("static source: load state: %w", err)
	}
	st := runtimeState{}
	if row.ExtraJSON != "" && row.ExtraJSON != "{}" {
		if err := json.Unmarshal([]byte(row.ExtraJSON), &st); err != nil {
			// The row is the source of truth for container_id /
			// proxy_route_id; only the optional state blob is at risk
			// and we fall back to zero state. Log so this is debuggable
			// after the fact.
			slog.Debug("static source: decode extra_json", "workload", w.ID, "error", err)
		}
	}
	return st, &row, nil
}

// saveLocks serializes per-workload read-modify-write of the container
// row inside saveState. With SQLite's MaxOpenConns=1, two parallel
// deploys for the same workload would still race the read+write
// against each other (the DB serializes statements but not the
// caller's intent), letting the loser's write stomp the winner's
// container_id / proxy_route_id and orphaning Docker resources. The
// mutex caps the concurrency at 1 per workload; cross-workload
// parallelism is unaffected.
//
// Entries are reference-counted and removed only when the last holder
// releases. This bounds memory (no per-workload-ID leak) WITHOUT the
// use-after-delete hazard of deleting an entry on teardown: deleting a
// live entry while a concurrent saveState still holds (or is about to
// lock) it would let a fresh saveState mint a SECOND mutex for the same
// workload, losing the RMW serialization the lock exists to provide.
var saveLocks struct {
	mu    sync.Mutex
	locks map[string]*saveLock
}

type saveLock struct {
	mu   sync.Mutex
	refs int
}

// acquireSaveLock returns the per-workload lock (creating it on first use),
// registers this caller as a holder, and takes the lock. Pair with
// releaseSaveLock. The outer mutex is held only for the bookkeeping; callers
// contend on the returned per-workload lock.
func acquireSaveLock(workloadID string) *saveLock {
	saveLocks.mu.Lock()
	if saveLocks.locks == nil {
		saveLocks.locks = map[string]*saveLock{}
	}
	l, ok := saveLocks.locks[workloadID]
	if !ok {
		l = &saveLock{}
		saveLocks.locks[workloadID] = l
	}
	l.refs++
	saveLocks.mu.Unlock()
	l.mu.Lock()
	return l
}

// releaseSaveLock unlocks and drops the caller's reference, removing the map
// entry once no holders remain. Because refs is incremented under saveLocks.mu
// before the entry can be observed for deletion, an entry with a pending
// acquirer is never deleted.
func releaseSaveLock(workloadID string, l *saveLock) {
	l.mu.Unlock()
	saveLocks.mu.Lock()
	l.refs--
	if l.refs == 0 {
		delete(saveLocks.locks, workloadID)
	}
	saveLocks.mu.Unlock()
}

// saveState upserts the container row, calling mutate so callers can
// adjust both the runtime state blob (extra_json) and the row's
// first-class fields (container_id, proxy_route_id, state, etc.) in a
// single transaction.
//
// The mutate callback receives a pointer to a runtimeState seeded from
// the existing extra_json; on save the typed fields are merged back on
// top of any unknown keys so future-writer values (e.g. per-face
// route maps) survive the round-trip.
//
// Per-workload mutex serializes concurrent callers so two parallel
// Deploys can't read the same prior state and race their writes.
func saveState(deps plugin.Deps, w plugin.Workload, mutate func(*runtimeState, *store.Container)) error {
	lk := acquireSaveLock(w.ID)
	defer releaseSaveLock(w.ID, lk)

	prev, prevRow, err := loadState(deps, w)
	if err != nil {
		return err
	}

	row := store.Container{
		ID:           containerRowID(w),
		WorkloadID:   w.ID,
		WorkloadKind: string(store.WorkloadKindSite),
		Host:         "local",
	}
	if prevRow != nil {
		row = *prevRow
	}

	// Round-trip extra_json through a generic map so unknown keys
	// survive. Strip the typed-state keys before the merge so that
	// a typed field cleared to its zero value (e.g. LastError = "")
	// actually drops the key instead of being re-introduced by the
	// generic decode.
	generic := map[string]json.RawMessage{}
	if row.ExtraJSON != "" && row.ExtraJSON != "{}" {
		if err := json.Unmarshal([]byte(row.ExtraJSON), &generic); err != nil {
			slog.Debug("static source: decode extra_json (generic)", "workload", w.ID, "error", err)
		}
	}
	for _, k := range runtimeStateKeys {
		delete(generic, k)
	}

	state := prev
	mutate(&state, &row)

	// Re-emit typed fields into the generic map so they win over any
	// historical key with the same name.
	typedBytes, err := json.Marshal(state)
	if err != nil {
		return fmt.Errorf("static source: marshal state: %w", err)
	}
	typedMap := map[string]json.RawMessage{}
	if err := json.Unmarshal(typedBytes, &typedMap); err != nil {
		return fmt.Errorf("static source: re-decode typed state: %w", err)
	}
	for k, v := range typedMap {
		generic[k] = v
	}

	merged, err := json.Marshal(generic)
	if err != nil {
		return fmt.Errorf("static source: marshal merged state: %w", err)
	}
	row.ExtraJSON = string(merged)
	row.LastSeenAt = store.Now()

	if err := deps.Store.UpsertContainer(row); err != nil {
		return fmt.Errorf("static source: upsert container row: %w", err)
	}
	return nil
}