EvoBGP/internal/store/memory_deployment.go

package store

import (
	"context"
	"sort"
	"time"
)

type deploymentReservation struct {
	Owner string
	Token int64
	Until time.Time
}

func (m *Memory) deploymentFence(f JobFence) (JobRecord, error) {
	j, ok := m.jobs[f.JobID]
	if !ok || j.TenantID != f.TenantID || j.Status != "running" || j.OwnerID != f.OwnerID || j.FencingToken != f.Token || j.CancelRequested || j.LeaseUntil == nil || !j.LeaseUntil.After(time.Now()) {
		return JobRecord{}, ErrJobLeaseLost
	}
	return j, nil
}

func (m *Memory) BeginDeploymentSpeaker(ctx context.Context, f JobFence, speaker, revision string) (DeploymentSpeaker, error) {
	if err := ctx.Err(); err != nil {
		return DeploymentSpeaker{}, err
	}
	m.jobMu.Lock()
	defer m.jobMu.Unlock()
	j, err := m.deploymentFence(f)
	if err != nil {
		return DeploymentSpeaker{}, err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	sp, ok := m.speakers[speaker]
	if !ok || sp.TenantID != f.TenantID {
		return DeploymentSpeaker{}, ErrNotFound
	}
	r, ok := m.revisions[revision]
	if !ok || r.TenantID != f.TenantID {
		return DeploymentSpeaker{}, ErrNotFound
	}
	if m.deployments == nil {
		m.deployments = make(map[string]DeploymentSpeaker)
		m.deploymentCurrent = make(map[string]DeploymentSpeaker)
	}
	key := f.JobID + ":" + speaker
	cur := m.deploymentCurrent[speaker]
	if f.ExpectedGeneration != 0 && cur.JobID != f.JobID && (cur.Generation != f.ExpectedGeneration || cur.RevisionID != revision) {
		return DeploymentSpeaker{}, ErrJobLeaseLost
	}
	reserve := func() error {
		if m.deploymentReservations == nil {
			m.deploymentReservations = make(map[string]deploymentReservation)
		}
		if r := m.deploymentReservations[speaker]; r.Until.After(time.Now()) {
			return ErrDeploymentBusy
		}
		m.deploymentReservations[speaker] = deploymentReservation{Owner: f.OwnerID, Token: f.Token, Until: time.Now().Add(DeploymentDeliveryWindow)}
		return nil
	}
	if existing, ok := m.deployments[key]; ok {
		if cur.JobID != f.JobID || cur.Generation != existing.Generation || cur.RevisionID != revision {
			return DeploymentSpeaker{}, ErrJobLeaseLost
		}
		if existing.State != "applied" {
			if err := reserve(); err != nil {
				return DeploymentSpeaker{}, err
			}
		}
		return existing, nil
	}
	if f.ExpectedGeneration != 0 {
		j.RequestSequence = cur.RequestSequence
	}
	if cur.RequestSequence >= j.RequestSequence && cur.JobID != "" && f.ExpectedGeneration == 0 {
		return DeploymentSpeaker{}, ErrJobLeaseLost
	}
	if err := reserve(); err != nil {
		return DeploymentSpeaker{}, err
	}
	d := DeploymentSpeaker{JobID: f.JobID, TenantID: f.TenantID, SpeakerID: speaker, RevisionID: revision, Generation: cur.Generation + 1, RequestSequence: j.RequestSequence, State: "published", RetryRound: f.RetryRound}
	m.deployments[key] = d
	m.deploymentCurrent[speaker] = d
	m.publishedRevision[speaker] = publishedInfo{RevisionID: revision, PublishedAt: time.Now().UTC()}
	return d, nil
}

func (m *Memory) RecordDeploymentSpeaker(ctx context.Context, f JobFence, d DeploymentSpeaker, state, confirmation, message string) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	m.jobMu.Lock()
	defer m.jobMu.Unlock()
	if _, err := m.deploymentFence(f); err != nil {
		return err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	cur := m.deploymentCurrent[d.SpeakerID]
	if d.JobID != f.JobID || d.TenantID != f.TenantID || cur.JobID != d.JobID || cur.Generation != d.Generation || cur.RevisionID != d.RevisionID || m.publishedRevision[d.SpeakerID].RevisionID != d.RevisionID {
		return ErrJobLeaseLost
	}
	r := m.deploymentReservations[d.SpeakerID]
	if r.Owner != f.OwnerID || r.Token != f.Token || !r.Until.After(time.Now()) {
		return ErrJobLeaseLost
	}
	if state == "applied" && confirmation != "agent_reported" && confirmation != "local_applied" {
		return ErrInvalidInput
	}
	if state != "dispatched" && state != "unknown" {
		delete(m.deploymentReservations, d.SpeakerID)
	}
	cur.State, cur.Confirmation, cur.Error = state, confirmation, message
	if state == "dispatched" {
		cur.DispatchAttempts++
	}
	if state == "applied" {
		revision := d.RevisionID
		m.speakers[d.SpeakerID].LastAppliedRevisionID = &revision
	}
	m.deploymentCurrent[d.SpeakerID] = cur
	m.deployments[d.JobID+":"+d.SpeakerID] = cur
	return nil
}

func (m *Memory) ListDeploymentSpeakers(ctx context.Context, tenant, job string) ([]DeploymentSpeaker, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	m.mu.RLock()
	defer m.mu.RUnlock()
	var out []DeploymentSpeaker
	for _, d := range m.deployments {
		if d.TenantID == tenant && d.JobID == job {
			out = append(out, d)
		}
	}
	return out, nil
}

func (m *Memory) ListPendingDeployments(ctx context.Context, limit int) ([]DeploymentSpeaker, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	m.jobMu.Lock()
	defer m.jobMu.Unlock()
	m.mu.RLock()
	defer m.mu.RUnlock()
	var out []DeploymentSpeaker
	for speaker, d := range m.deploymentCurrent {
		j := m.jobs[d.JobID]
		if d.State != "applied" && d.RetryRound < 3 && !j.CancelRequested && (j.Status == "failed" || j.Status == "succeeded") && !m.deploymentReservations[speaker].Until.After(time.Now()) {
			out = append(out, d)
			if len(out) >= limit {
				break
			}
		}
	}
	return out, nil
}

func (m *Memory) ListDeploymentObservations(ctx context.Context, limit int) ([]DeploymentSpeaker, error) {
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	if limit <= 0 || limit > 100 {
		limit = 10
	}
	m.jobMu.Lock()
	defer m.jobMu.Unlock()
	m.mu.Lock()
	defer m.mu.Unlock()
	var out []DeploymentSpeaker
	for speaker, d := range m.deploymentCurrent {
		j := m.jobs[d.JobID]
		if !j.CancelRequested && (j.Status == "failed" || j.Status == "succeeded") && !m.deploymentReservations[speaker].Until.After(time.Now()) {
			out = append(out, d)
		}
	}
	sort.Slice(out, func(i, j int) bool {
		a, b := out[i].HealthReceivedAt, out[j].HealthReceivedAt
		if a == nil && b == nil {
			return out[i].SpeakerID < out[j].SpeakerID
		}
		if a == nil {
			return true
		}
		if b == nil {
			return false
		}
		return a.Before(*b)
	})
	if len(out) > limit {
		out = out[:limit]
	}
	for i := range out {
		out[i].ObservationToken++
		m.deploymentCurrent[out[i].SpeakerID] = out[i]
		m.deployments[out[i].JobID+":"+out[i].SpeakerID] = out[i]
	}
	return out, nil
}

func (m *Memory) ObserveDeployment(ctx context.Context, d DeploymentSpeaker, o DeploymentObservation) error {
	if err := ctx.Err(); err != nil {
		return err
	}
	m.mu.Lock()
	defer m.mu.Unlock()
	cur := m.deploymentCurrent[d.SpeakerID]
	if cur.TenantID != d.TenantID || cur.JobID != d.JobID || cur.Generation != d.Generation || cur.RevisionID != d.RevisionID || cur.ObservationToken != d.ObservationToken || m.publishedRevision[d.SpeakerID].RevisionID != d.RevisionID || m.deploymentReservations[d.SpeakerID].Until.After(time.Now()) {
		return ErrJobLeaseLost
	}
	now := time.Now().UTC()
	cur.ObservationToken++
	cur.HealthReceivedAt = &now
	cur.HealthRevisionID = o.RevisionID
	cur.HealthSourceAt = o.SourceAt
	cur.HealthError = o.Error
	if o.Error == "" && o.RevisionID != "" {
		if o.RevisionID == d.RevisionID {
			cur.State = "applied"
			cur.Confirmation = "agent_reported"
			revision := d.RevisionID
			m.speakers[d.SpeakerID].LastAppliedRevisionID = &revision
		} else {
			cur.State = "unknown"
			cur.Confirmation = "unconfirmed"
		}
	}
	m.deploymentCurrent[d.SpeakerID] = cur
	m.deployments[d.JobID+":"+d.SpeakerID] = cur
	return nil
}