bri-sandbox-development-platform/control-plane/internal/store/deployments.go

package store

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"os"
	"path/filepath"
	"time"

	"github.com/sdp/protocol"
)

// Deployment is a deployment row joined with the current state and
// container info. Used by the dashboard's history list and the
// "active deployment" badge in the sandbox view.
type Deployment struct {
	ID          string `json:"id"`
	SandboxID   string `json:"sandboxId,omitempty"`
	Repository  string `json:"repository"`
	Branch      string `json:"branch"`
	User        string `json:"user"`
	State       string `json:"state"`
	ContainerID string `json:"containerId,omitempty"`
	HostPort    int    `json:"hostPort,omitempty"`
	StartedAt   int64  `json:"startedAt"`
	CompletedAt int64  `json:"completedAt,omitempty"`
}

// StartDeployment records a new deployment row. Idempotent on id.
func (s *Store) StartDeployment(id, repo, branch, user string) error {
	_, err := s.db.Exec(
		`INSERT OR IGNORE INTO deployments(id, repository, branch, user, state, started_at) VALUES(?,?,?,?,?,?)`,
		id, repo, branch, user, "QUEUED", time.Now().UnixMilli(),
	)
	return err
}

// StartDeploymentInSandbox records a new deployment tied to a sandbox.
// sandboxID and hostPort are nullable; empty string / 0 stores NULL.
func (s *Store) StartDeploymentInSandbox(id, sandboxID, repo, branch, user string, hostPort int) error {
	var sb interface{}
	if sandboxID != "" {
		sb = sandboxID
	}
	var hp interface{}
	if hostPort > 0 {
		hp = hostPort
	}
	_, err := s.db.Exec(
		`INSERT OR IGNORE INTO deployments(id, sandbox_id, repository, branch, user, state, host_port, started_at)
		 VALUES(?,?,?,?,?,?,?,?)`,
		id, sb, repo, branch, user, "QUEUED", hp, time.Now().UnixMilli(),
	)
	return err
}

// FinishDeployment marks the final state. completed_at is set if state is terminal.
func (s *Store) FinishDeployment(id, state string) error {
	_, err := s.db.Exec(
		`UPDATE deployments SET state=?, completed_at=? WHERE id=?`,
		state, time.Now().UnixMilli(), id,
	)
	return err
}

// SetContainerID records the container id once a deploy is RUNNING.
func (s *Store) SetContainerID(id, containerID string) error {
	_, err := s.db.Exec(`UPDATE deployments SET container_id=? WHERE id=?`, containerID, id)
	return err
}

// MarkStage records a stage transition. ok=1 success, 0 failure.
func (s *Store) MarkStage(id, stage string, ok bool) error {
	v := 0
	if ok {
		v = 1
	}
	_, err := s.db.Exec(
		`INSERT INTO progress(deployment_id, stage, ok, at) VALUES(?,?,?,?)`,
		id, stage, v, time.Now().UnixMilli(),
	)
	return err
}

// AppendEvent writes an event. Log lines go to .log; progress/status hit SQLite.
// The deployment's running state is also updated so /api/deployments/{id} can
// serve a snapshot without replaying the whole log.
func (s *Store) AppendEvent(e protocol.Event) error {
	switch e.Kind {
	case "log":
		return s.appendLog(e)
	case "status":
		_, err := s.db.Exec(`UPDATE deployments SET state=? WHERE id=?`, e.State, e.DeploymentID)
		return err
	case "progress":
		return s.MarkStage(e.DeploymentID, e.Stage, e.State != "FAILED")
	}
	return nil
}

func (s *Store) appendLog(e protocol.Event) error {
	s.mu.Lock()
	f, ok := s.logs[e.DeploymentID]
	if !ok {
		path := filepath.Join(s.dir, "logs", e.DeploymentID+".log")
		var err error
		f, err = os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
		if err != nil {
			s.mu.Unlock()
			return err
		}
		s.logs[e.DeploymentID] = f
	}
	s.mu.Unlock()
	ts := time.UnixMilli(e.At).Format("15:04:05.000")
	_, err := fmt.Fprintf(f, "%s %s\n", ts, e.Line)
	return err
}

// TailLogs returns the last n lines of a deployment's log file. Used by the
// dashboard on first connect to backfill.
func (s *Store) TailLogs(id string, n int) ([]string, error) {
	path := filepath.Join(s.dir, "logs", id+".log")
	data, err := os.ReadFile(path)
	if err != nil {
		if os.IsNotExist(err) {
			return nil, nil
		}
		return nil, err
	}
	// ponytail: O(n) scan, fine for tail use. swap to a ring buffer if logs get huge.
	var lines []string
	start := 0
	for i, b := range data {
		if b == '\n' {
			lines = append(lines, string(data[start:i]))
			start = i + 1
		}
	}
	if start < len(data) {
		lines = append(lines, string(data[start:]))
	}
	if n > 0 && len(lines) > n {
		lines = lines[len(lines)-n:]
	}
	return lines, nil
}

// ListDeployments returns deployments, newest first. If sandboxID is set,
// filters to that sandbox. limit caps the result.
func (s *Store) ListDeployments(sandboxID string, limit int) ([]Deployment, error) {
	if limit <= 0 || limit > 500 {
		limit = 100
	}
	var (
		rows *sql.Rows
		err  error
	)
	if sandboxID == "" {
		rows, err = s.db.Query(
			`SELECT id, COALESCE(sandbox_id,''), repository, branch, COALESCE(user,''), state,
			        COALESCE(container_id,''), COALESCE(host_port,0), started_at, COALESCE(completed_at,0)
			 FROM deployments ORDER BY started_at DESC LIMIT ?`, limit)
	} else {
		rows, err = s.db.Query(
			`SELECT id, COALESCE(sandbox_id,''), repository, branch, COALESCE(user,''), state,
			        COALESCE(container_id,''), COALESCE(host_port,0), started_at, COALESCE(completed_at,0)
			 FROM deployments WHERE sandbox_id=? ORDER BY started_at DESC LIMIT ?`, sandboxID, limit)
	}
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Deployment
	for rows.Next() {
		var d Deployment
		if err := rows.Scan(&d.ID, &d.SandboxID, &d.Repository, &d.Branch, &d.User, &d.State,
			&d.ContainerID, &d.HostPort, &d.StartedAt, &d.CompletedAt); err != nil {
			return nil, err
		}
		out = append(out, d)
	}
	return out, rows.Err()
}

// LatestDeploymentBySandboxService returns the most recent deployment for
// the (sandbox, repo) pair, or nil if none.
func (s *Store) LatestDeploymentBySandboxService(sandboxID, repo string) (*Deployment, error) {
	row := s.db.QueryRow(
		`SELECT id, COALESCE(sandbox_id,''), repository, branch, COALESCE(user,''), state,
		        COALESCE(container_id,''), COALESCE(host_port,0), started_at, COALESCE(completed_at,0)
		 FROM deployments WHERE sandbox_id=? AND repository=? ORDER BY started_at DESC LIMIT 1`,
		sandboxID, repo)
	var d Deployment
	err := row.Scan(&d.ID, &d.SandboxID, &d.Repository, &d.Branch, &d.User, &d.State,
		&d.ContainerID, &d.HostPort, &d.StartedAt, &d.CompletedAt)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &d, nil
}

// GetDeployment returns one deployment by id, or nil if not found.
func (s *Store) GetDeployment(id string) (*Deployment, error) {
	row := s.db.QueryRow(
		`SELECT id, COALESCE(sandbox_id,''), repository, branch, COALESCE(user,''), state,
		        COALESCE(container_id,''), COALESCE(host_port,0), started_at, COALESCE(completed_at,0)
		 FROM deployments WHERE id=?`, id)
	var d Deployment
	err := row.Scan(&d.ID, &d.SandboxID, &d.Repository, &d.Branch, &d.User, &d.State,
		&d.ContainerID, &d.HostPort, &d.StartedAt, &d.CompletedAt)
	if err == sql.ErrNoRows {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &d, nil
}

// marshalJSON is a small helper for the json column on environments.
func marshalJSON(v map[string]string) (string, error) {
	b, err := json.Marshal(v)
	if err != nil {
		return "", err
	}
	return string(b), nil
}

func unmarshalJSON(s string) (map[string]string, error) {
	if s == "" {
		return map[string]string{}, nil
	}
	var m map[string]string
	if err := json.Unmarshal([]byte(s), &m); err != nil {
		return nil, err
	}
	return m, nil
}