dsh-web/internal/api/images_policy.go

package api

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"git.misaka.ren/m1saka/dsh-web/internal/docker"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"runtime"
	"syscall"
	"time"
)

const imageRecipeRev = "7"

var imagePlatform = runtime.GOOS + "/" + runtime.GOARCH

const minimumBuildDiskBytes = uint64(2 << 30)

var buildDiskAvailable = func(path string) (uint64, error) {
	var stat syscall.Statfs_t
	if err := syscall.Statfs(path, &stat); err != nil {
		return 0, err
	}
	return uint64(stat.Bavail) * uint64(stat.Bsize), nil
}

type inspectedImage struct {
	ID                                 string `json:"Id"`
	Os, Architecture, Variant, Created string
	RepoTags                           []string
	Config                             struct{ Labels map[string]string }
}

func (im inspectedImage) platform() string {
	if im.Os == "" || im.Architecture == "" {
		return ""
	}
	p := im.Os + "/" + im.Architecture
	// Docker can report the default ARM64 variant explicitly. Keep its
	// identity consistent with build records and admission keys; preserve
	// all other variants so incompatible images remain rejected.
	if im.Architecture == "arm64" && im.Variant == "v8" {
		return p
	}
	if im.Variant != "" {
		p += "/" + im.Variant
	}
	return p
}

func (a *API) inspectLocalImage(ctx context.Context, id string) (*inspectedImage, error) {
	if id == "" {
		return nil, nil
	}
	var im inspectedImage
	if err := a.docker.Do(ctx, http.MethodGet, "/images/"+url.PathEscape(id)+"/json", nil, &im); err != nil {
		var apiErr *docker.APIError
		if errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusNotFound {
			return nil, nil
		}
		return nil, err
	}
	return &im, nil
}

func (a *API) imageMatches(ctx context.Context, id, version string) (bool, error) {
	im, err := a.inspectLocalImage(ctx, id)
	if err != nil || im == nil {
		return false, err
	}
	return im.ID == id && im.platform() == imagePlatform && im.Config.Labels["dsh-mgr.role"] == "runtime" && im.Config.Labels["dsh-mgr.version"] == version && im.Config.Labels["dsh-mgr.recipe"] == imageRecipeRev, nil
}

var errImageIndexChanged = errors.New("image inventory changed during reconciliation")

func (a *API) imageIndex(ctx context.Context) ([]imageRecord, error) {
	// Retry stale observations without holding the queue across Docker I/O.
	// Bound retries so sustained deletions cannot keep a request alive forever.
	for attempt := 0; attempt < 3; attempt++ {
		result, err := a.imageIndexSnapshot(ctx)
		if !errors.Is(err, errImageIndexChanged) {
			return result, err
		}
	}
	return nil, errImageIndexChanged
}

func (a *API) imageIndexSnapshot(ctx context.Context) ([]imageRecord, error) {
	// Phase 1: read the managed index under the queue lock (DB only).
	buildQueue.Lock()
	generation := a.imageDeleteGeneration
	err := a.ensureBuilds()
	if err == nil {
		_, err = a.s.DB.ExecContext(ctx, `INSERT OR IGNORE INTO images(image_id,version,recipe_rev,platform,tag,status,built_at)
		SELECT image_id,version,recipe_rev,platform,'','missing',COALESCE(finished_at,created_at) FROM image_builds WHERE status='success' AND image_id<>'' ORDER BY created_at DESC,id DESC`)
	}
	var rows *sql.Rows
	if err == nil {
		rows, err = a.s.DB.QueryContext(ctx, `SELECT image_id,version,COALESCE(recipe_rev,''),COALESCE(platform,''),tag,status,built_at,
		(SELECT count(*) FROM instances WHERE instances.image_id=images.image_id) FROM images ORDER BY built_at DESC,image_id`)
	}
	result := []imageRecord{}
	if err == nil {
		for rows.Next() {
			var im imageRecord
			if err = rows.Scan(&im.ImageID, &im.Version, &im.RecipeRev, &im.Platform, &im.Tag, &im.Status, &im.Created, &im.UsageCount); err != nil {
				break
			}
			im.ID = im.ImageID
			result = append(result, im)
		}
		scanErr := rows.Err()
		rows.Close()
		if err == nil {
			err = scanErr
		}
	}
	buildQueue.Unlock()
	if err != nil {
		return nil, err
	}
	// Phase 2: inspect Docker without the queue lock; per-image state comes
	// from real daemon observations, and instance creation must not wait here.
	known := make(map[string]bool, len(result))
	for _, im := range result {
		known[im.ImageID] = true
	}
	var local []struct {
		ID     string
		Labels map[string]string
	}
	{
		filters, _ := json.Marshal(map[string][]string{"label": {"dsh-mgr.role=runtime"}})
		if err := a.docker.Do(ctx, http.MethodGet, "/images/json?filters="+url.QueryEscape(string(filters)), nil, &local); err != nil {
			return nil, err
		}
	}
	for _, im := range local {
		if im.ID != "" && !known[im.ID] && im.Labels["dsh-mgr.role"] == "runtime" {
			known[im.ID] = true
			result = append(result, imageRecord{ID: im.ID, ImageID: im.ID})
		}
	}
	for i := range result {
		im := &result[i]
		im.Status = "missing"
		im.Admission = "existing_only"
		im.AdmissionReason = "local image missing"
		actual, err := a.inspectLocalImage(ctx, im.ImageID)
		if err != nil {
			return nil, err
		}
		if actual == nil {
			continue
		}
		im.Status = "unverified"
		im.Platform = actual.platform()
		im.Version = actual.Config.Labels["dsh-mgr.version"]
		im.RecipeRev = actual.Config.Labels["dsh-mgr.recipe"]
		im.Tag = ""
		if len(actual.RepoTags) > 0 {
			im.Tag = actual.RepoTags[0]
		}
		if actual.Created != "" {
			im.Created = actual.Created
		}
		if err := a.s.DB.QueryRowContext(ctx, "SELECT count(*) FROM instances WHERE image_id=?", im.ImageID).Scan(&im.UsageCount); err != nil {
			return nil, err
		}
		policy, err := a.evaluateImage(ctx, im.ImageID)
		if err != nil {
			return nil, err
		}
		im.Status, im.Admission, im.AdmissionReason, im.Selectable = policy.Status, policy.Admission, policy.AdmissionReason, policy.Selectable
	}
	// Phase 3: write observed state back under the queue lock (DB only).
	buildQueue.Lock()
	defer buildQueue.Unlock()
	if generation != a.imageDeleteGeneration {
		return nil, errImageIndexChanged
	}
	tx, err := a.s.DB.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	present := make([]imageRecord, 0, len(result))
	for _, im := range result {
		if im.Status == "missing" {
			if _, err = tx.ExecContext(ctx, "UPDATE image_builds SET status='missing',error='image no longer available or template identity is unknown' WHERE image_id=? AND status='success'", im.ImageID); err != nil {
				return nil, err
			}
			if _, err = tx.ExecContext(ctx, "DELETE FROM images WHERE image_id=?", im.ImageID); err != nil {
				return nil, err
			}
			continue
		}
		if _, err = tx.ExecContext(ctx, `INSERT INTO images(image_id,version,recipe_rev,platform,tag,status,built_at,usage_count) VALUES(?,?,?,?,?,?,?,?)
			ON CONFLICT(image_id) DO UPDATE SET version=excluded.version,recipe_rev=excluded.recipe_rev,platform=excluded.platform,tag=excluded.tag,status=excluded.status,built_at=excluded.built_at,usage_count=excluded.usage_count`, im.ImageID, im.Version, im.RecipeRev, im.Platform, im.Tag, im.Status, im.Created, im.UsageCount); err != nil {
			return nil, err
		}
		present = append(present, im)
	}
	if err = tx.Commit(); err != nil {
		return nil, err
	}
	return present, nil
}

func (a *API) StartImageBuilds() error {
	buildQueue.Lock()
	defer buildQueue.Unlock()
	if err := a.ensureBuilds(); err != nil {
		return err
	}
	a.startBuildWorkerLocked()
	return nil
}

func (a *API) checkBuildDisk(ctx context.Context) error {
	var info struct{ DockerRootDir string }
	if err := a.docker.Do(ctx, http.MethodGet, "/info", nil, &info); err != nil {
		return fmt.Errorf("build disk check: %w", err)
	}
	// A matching pathname alone cannot establish daemon filesystem identity.
	confirmed := os.Getenv("DSH_DOCKER_ROOT_VISIBLE")
	if !filepath.IsAbs(info.DockerRootDir) || confirmed != info.DockerRootDir {
		return errors.New("Docker disk capacity unknown: mount the daemon DockerRootDir at the identical path and set DSH_DOCKER_ROOT_VISIBLE to that path; new builds are blocked")
	}
	available, err := buildDiskAvailable(confirmed)
	if err != nil {
		return fmt.Errorf("Docker disk capacity unavailable: %w", err)
	}
	return requireBuildDisk(available)
}

func requireBuildDisk(available uint64) error {
	if available < minimumBuildDiskBytes {
		return fmt.Errorf("insufficient Docker disk space: %d available bytes, require %d", available, minimumBuildDiskBytes)
	}
	return nil
}

func (a *API) insertBuildRequest(b buildRecord, user string) error {
	tx, err := a.s.DB.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if _, err = tx.Exec("INSERT INTO image_builds(id,version,status,created_at,requested_by,recipe_rev,platform,experimental,release_id,source_repository,source_commit) VALUES(?,?,?,?,?,?,?,?,?,?,?)", b.ID, b.Version, b.Status, b.Created, b.RequestedBy, b.RecipeRev, b.Platform, b.Experimental, b.ReleaseID, b.SourceRepository, b.SourceCommit); err != nil {
		return err
	}
	if _, err = tx.Exec("INSERT INTO image_build_requests(user_id,build_id,created_at) VALUES(?,?,?)", user, b.ID, b.Created); err != nil {
		return err
	}
	return tx.Commit()
}

func (a *API) reconcileBuilds(parent context.Context, records []buildRecord) error {
	ctx, cancel := context.WithTimeout(parent, 15*time.Second)
	defer cancel()
	for i := range records {
		b := &records[i]
		if b.Status != "success" {
			continue
		}
		actual, err := a.inspectLocalImage(ctx, b.ImageID)
		if err != nil {
			return err
		}
		if actual == nil || actual.ID != b.ImageID || actual.Config.Labels["dsh-mgr.role"] != "runtime" || actual.Config.Labels["dsh-mgr.version"] != b.Version || actual.Config.Labels["dsh-mgr.recipe"] != b.RecipeRev || actual.platform() != b.Platform || !releaseLabelsMatch(actual, *b) {
			b.Status, b.Error = "missing", "image no longer available"
			if _, err := a.s.DB.Exec("UPDATE image_builds SET status=?,error=? WHERE id=? AND status='success'", b.Status, b.Error, b.ID); err != nil {
				return err
			}
		}
	}
	return nil
}