pages/internal/deploy/service.go

package deploy

import (
	"context"
	"errors"
	"io"
	"log/slog"
	"sync"
	"time"

	"github.com/iceBear67/simplepages/api"
	"github.com/iceBear67/simplepages/internal/cas"
	"github.com/iceBear67/simplepages/internal/site"
	"github.com/iceBear67/simplepages/internal/store"
	"github.com/iceBear67/simplepages/internal/webroot"
)

// Service runs the deployment lifecycle.
//
// Client-visible failures are returned as *api.Error so the HTTP layer stays a
// translation of shapes rather than a second copy of the rules; anything else
// is an internal error and is rendered as an opaque 500 by httpx.WriteError.
type Service struct {
	DB  *store.DB
	CAS *cas.Store
	Log *slog.Logger

	// Dir is the root of the assembled deployment trees. Empty means the
	// operator chose assemble_mode=none: content is served straight from the
	// CAS and nothing is built on disk.
	Dir string

	// Sites is the in-memory state the HTTP site handler reads. Activation
	// publishes into it; nil leaves the service usable without a serving layer,
	// which is what the store-level tests want.
	Sites *site.Registry

	// Webroot maintains the $WEBROOT/~project symlinks. Nil when the operator
	// configured no webroot. Nothing here depends on it succeeding.
	Webroot *webroot.Webroot

	// BlobGrace is how long content must have been unreferenced before the
	// collector removes it. Zero means defaultBlobGrace. Negative collects
	// immediately, which is what the tests want and what an operator reclaiming
	// space on a server they know is idle might ask for.
	BlobGrace time.Duration

	locks projectLocks
}

// Create starts a deployment. Nothing touches the filesystem until a manifest
// arrives, so an abandoned create costs one row.
func (s *Service) Create(ctx context.Context, p *store.Project, keyID string, meta map[string]string) (*store.Deployment, error) {
	dep := &store.Deployment{ProjectID: p.ID, CreatedByKey: keyID, Meta: meta}
	if err := s.DB.CreateDeployment(ctx, dep); err != nil {
		return nil, err
	}
	return dep, nil
}

// SetManifest records the file list and reports which blobs still have to be
// uploaded. files must already be validated: paths through pathutil and sizes
// against the project's limits.
func (s *Service) SetManifest(ctx context.Context, dep *store.Deployment, files []store.FileRow) (missing []cas.Digest, missingBytes int64, err error) {
	missing, missingBytes, err = s.DB.SetManifest(ctx, dep.ID, files)
	if err != nil {
		switch {
		case errors.Is(err, store.ErrConflict):
			return nil, 0, api.Errorf(api.CodeConflict,
				"this deployment can no longer accept a manifest; create a new one")
		case errors.Is(err, cas.ErrSizeMismatch):
			return nil, 0, api.Errorf(api.CodeSizeMismatch, "%s", err)
		}
		return nil, 0, err
	}
	return missing, missingBytes, nil
}

// Upload stores one blob's content.
//
// The digest must already be named by some manifest. That check is what keeps
// the endpoint from being general-purpose storage: content nobody declared can
// never be written, and the length it must have is the one the manifest agreed
// on rather than whatever Content-Length claims.
//
// Reports whether the content was newly stored; a blob that is already present
// is a success without reading the body, which is what makes a retried deploy
// cheap.
func (s *Service) Upload(ctx context.Context, digest cas.Digest, body io.Reader) (size int64, stored bool, err error) {
	b, err := s.DB.Blob(ctx, digest)
	if err != nil {
		if errors.Is(err, store.ErrNotFound) {
			return 0, false, api.Errorf(api.CodeNotFound,
				"no manifest references this digest; send the manifest first")
		}
		return 0, false, err
	}
	if b.Present {
		return b.Size, false, nil
	}

	// The blob's declared length is both the expectation and the ceiling: Put
	// reads one byte past it and rejects anything longer, so a client cannot
	// spend more of the disk than its manifest was accepted for. Put requires a
	// positive ceiling, hence the floor of one byte for an empty blob.
	limit := b.Size
	if limit < 1 {
		limit = 1
	}
	n, err := s.CAS.Put(ctx, digest, b.Size, limit, body)
	if err != nil {
		switch {
		case errors.Is(err, cas.ErrDigestMismatch):
			return 0, false, api.Errorf(api.CodeDigestMismatch, "%s", err)
		case errors.Is(err, cas.ErrSizeMismatch):
			return 0, false, api.Errorf(api.CodeSizeMismatch, "%s", err)
		case errors.Is(err, cas.ErrTooLarge):
			return 0, false, api.Errorf(api.CodeLimitExceeded, "%s", err)
		}
		return 0, false, err
	}
	if err := s.DB.MarkBlobPresent(ctx, digest, n); err != nil {
		// The content is on disk and verified; only the row disagrees. A retry
		// finds the blob already stored and updates the row then.
		return 0, false, err
	}
	return n, true, nil
}

// Finalize checks that every blob arrived, assembles the tree, and marks the
// deployment ready. It does not activate it: a ready deployment is one that
// could be served, and choosing when to serve it is a separate decision.
//
// Retrying is safe. An assembled tree is left as it is, and a deployment that
// is already ready simply stays ready.
func (s *Service) Finalize(ctx context.Context, p *store.Project, dep *store.Deployment) (*store.Deployment, error) {
	// One finalize per project at a time. Two concurrent CI jobs for one
	// project are ordered rather than racing over the same directory.
	unlock, err := s.locks.lock(ctx, p.ID)
	if err != nil {
		return nil, err
	}
	defer unlock()

	// Re-read under the lock: the state may have moved since the handler
	// resolved it.
	dep, err = s.DB.DeploymentByPublicID(ctx, p.ID, dep.PublicID)
	if err != nil {
		return nil, mapNotFound(err)
	}
	switch dep.State {
	case store.StateUploading, store.StateReady:
	default:
		return nil, api.Errorf(api.CodeConflict, "cannot finalize a deployment that is %s", dep.State)
	}

	missing, err := s.DB.MissingBlobs(ctx, dep.ID)
	if err != nil {
		return nil, err
	}
	if len(missing) > 0 {
		hex := make([]string, len(missing))
		for i, d := range missing {
			hex[i] = d.String()
		}
		return nil, api.Errorf(api.CodeBlobsMissing,
			"%d blobs have not been uploaded", len(missing)).WithDetail("missing", hex)
	}

	if s.Dir != "" {
		files, err := s.DB.DeploymentFiles(ctx, dep.ID)
		if err != nil {
			return nil, err
		}
		if err := Assemble(ctx, s.CAS, files, DeploymentDir(s.Dir, p.ID, dep.PublicID)); err != nil {
			// A cancelled request is not a broken deployment: leave it uploading
			// so the client can simply try again.
			if ctx.Err() != nil {
				return nil, err
			}
			s.Log.ErrorContext(ctx, "assembling deployment tree failed",
				"project", p.Name, "deployment", dep.PublicID, "err", err)
			if ferr := s.DB.MarkDeploymentFailed(context.WithoutCancel(ctx), dep.ID, err.Error()); ferr != nil {
				s.Log.ErrorContext(ctx, "recording the failure failed too",
					"deployment", dep.PublicID, "err", ferr)
			}
			return nil, api.Errorf(api.CodeInternal, "could not assemble the deployment tree")
		}
	}

	if err := s.DB.MarkDeploymentReady(ctx, dep.ID); err != nil {
		if errors.Is(err, store.ErrConflict) {
			return nil, api.Errorf(api.CodeConflict, "%s", err)
		}
		return nil, err
	}
	return s.DB.DeploymentByPublicID(ctx, p.ID, dep.PublicID)
}

// Activate makes a ready deployment the one the project serves. Activating an
// older deployment is how a rollback works, and costs exactly the same.
//
// The order of the steps is the entire correctness argument:
//
//  1. Take the project lock, so two activations of one project are ordered.
//  2. Re-read the deployment under it and require that it is ready.
//  3. Build the snapshot's index — the one step that reads the manifest and can
//     fail — *before* anything has changed. A failure here leaves the currently
//     served deployment exactly as it was.
//  4. Commit the database transaction. From here on SQLite is the truth.
//  5. Store the pointer. This single store is the switch: requests that started
//     earlier finish on the old snapshot, later ones see the new one, and no
//     request can ever observe a mixture of the two.
//  6. Repoint the symlink, best effort.
//
// Database before memory matters: a crash between 4 and 5 restarts into a
// process that serves what the database says. The reverse order would leave a
// process serving something the database disagrees with.
func (s *Service) Activate(ctx context.Context, p *store.Project, dep *store.Deployment) (*store.Deployment, error) {
	unlock, err := s.locks.lock(ctx, p.ID)
	if err != nil {
		return nil, err
	}
	defer unlock()

	dep, err = s.DB.DeploymentByPublicID(ctx, p.ID, dep.PublicID)
	if err != nil {
		return nil, mapNotFound(err)
	}
	if dep.State != store.StateReady {
		return nil, api.Errorf(api.CodeDeploymentNotReady,
			"cannot activate a deployment that is %s", dep.State)
	}

	idx, err := s.index(ctx, dep)
	if err != nil {
		return nil, err
	}

	if err := s.DB.ActivateDeployment(ctx, p.ID, dep.ID); err != nil {
		switch {
		case errors.Is(err, store.ErrNotFound):
			return nil, api.Errorf(api.CodeNotFound, "no such deployment")
		case errors.Is(err, store.ErrConflict):
			return nil, api.Errorf(api.CodeConflict, "%s", err)
		}
		return nil, err
	}

	// Re-read so the snapshot and the response carry the timestamps the
	// transaction actually wrote.
	dep, err = s.DB.DeploymentByPublicID(ctx, p.ID, dep.PublicID)
	if err != nil {
		return nil, mapNotFound(err)
	}

	dir := s.deploymentDir(p, dep)
	if s.Sites != nil {
		s.Sites.Put(p).Activate(site.NewDeployment(dep, idx, dir))
	}
	if s.Webroot != nil && dir != "" {
		if err := s.Webroot.Point(p.Name, dir); err != nil {
			// The site is already being served from memory; the symlink is for
			// everything else and the reconciler will fix it.
			s.Log.ErrorContext(ctx, "could not repoint the webroot symlink",
				"project", p.Name, "deployment", dep.PublicID, "err", err)
		}
	}
	s.Log.InfoContext(ctx, "deployment activated",
		"project", p.Name, "deployment", dep.PublicID,
		"files", dep.FileCount, "bytes", dep.TotalBytes)
	return dep, nil
}

func mapNotFound(err error) error {
	if errors.Is(err, store.ErrNotFound) {
		return api.Errorf(api.CodeNotFound, "no such deployment")
	}
	return err
}

// projectLocks serialises the mutating operations of one project against each
// other. Entries are keyed by row id and are never evicted: there is one per
// project that has ever been written to, which is bounded by the number of
// projects.
type projectLocks struct {
	mu sync.Mutex
	m  map[int64]chan struct{}
}

// lock acquires the project's lock, or gives up if ctx is done first — a
// client that has already hung up should not keep a slow assembly waiting.
func (l *projectLocks) lock(ctx context.Context, id int64) (func(), error) {
	l.mu.Lock()
	if l.m == nil {
		l.m = make(map[int64]chan struct{})
	}
	ch, ok := l.m[id]
	if !ok {
		ch = make(chan struct{}, 1)
		l.m[id] = ch
	}
	l.mu.Unlock()

	select {
	case ch <- struct{}{}:
		return func() { <-ch }, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}