teraWeb/backend/internal/alarmdelivery/dispatcher.go

package alarmdelivery

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"

	"tera-clone/services/tera-api/internal/storage/atomicfile"
)

const defaultMaxJobs = 256

const maxStateBytes = 16 * 1024 * 1024

type DispatcherOptions struct {
	StatePath    string
	MaxJobs      int
	Clock        func() time.Time
	PollInterval time.Duration
}

type Dispatcher struct {
	statePath    string
	maxJobs      int
	clock        func() time.Time
	pollInterval time.Duration
	transports   map[string]Transport
	jobs         map[string]deliveryJob
	sequence     int
	ctx          context.Context
	cancel       context.CancelFunc
	wake         chan struct{}
	wg           sync.WaitGroup
	closeOnce    sync.Once
	mu           sync.Mutex
}

type deliveryJob struct {
	DeliveryID  string     `json:"delivery_id"`
	State       string     `json:"state"`
	Request     Request    `json:"request"`
	Attempts    int        `json:"attempts"`
	LastError   string     `json:"last_error,omitempty"`
	AcceptedAt  time.Time  `json:"accepted_at"`
	UpdatedAt   time.Time  `json:"updated_at"`
	NextAttempt time.Time  `json:"next_attempt"`
	CompletedAt *time.Time `json:"completed_at,omitempty"`
}

type dispatcherSnapshot struct {
	FormatVersion int                    `json:"format_version"`
	Sequence      int                    `json:"sequence"`
	Jobs          map[string]deliveryJob `json:"jobs"`
}

func NewDispatcher(transports map[string]Transport, options DispatcherOptions) (*Dispatcher, error) {
	if options.StatePath == "" || !filepath.IsAbs(options.StatePath) {
		return nil, errors.New("alarm delivery state path must be absolute")
	}
	if options.MaxJobs == 0 {
		options.MaxJobs = defaultMaxJobs
	}
	if options.MaxJobs < 1 || options.MaxJobs > 4096 {
		return nil, errors.New("alarm delivery max jobs must be between 1 and 4096")
	}
	if options.Clock == nil {
		options.Clock = time.Now
	}
	if options.PollInterval <= 0 {
		options.PollInterval = 250 * time.Millisecond
	}
	if options.PollInterval < 10*time.Millisecond || options.PollInterval > time.Second {
		return nil, errors.New("alarm delivery poll interval must be between 10ms and 1s")
	}
	if err := ensureStateDirectory(filepath.Dir(options.StatePath)); err != nil {
		return nil, err
	}
	ctx, cancel := context.WithCancel(context.Background())
	dispatcher := &Dispatcher{
		statePath: filepath.Clean(options.StatePath), maxJobs: options.MaxJobs,
		clock: options.Clock, pollInterval: options.PollInterval,
		transports: make(map[string]Transport), jobs: make(map[string]deliveryJob),
		ctx: ctx, cancel: cancel, wake: make(chan struct{}, 1),
	}
	for channel, transport := range transports {
		if transport != nil {
			dispatcher.transports[channel] = transport
		}
	}
	if err := dispatcher.load(); err != nil {
		cancel()
		return nil, err
	}
	dispatcher.wg.Add(1)
	go dispatcher.worker()
	dispatcher.notify()
	return dispatcher, nil
}

func (d *Dispatcher) Deliver(ctx context.Context, request Request) (Result, error) {
	request, err := NormalizeRequest(request)
	if err != nil {
		return Result{}, err
	}
	if err := ctx.Err(); err != nil {
		return Result{}, err
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	if d.ctx.Err() != nil {
		return Result{}, &UnavailableError{Reason: "tera-alarmd is stopping"}
	}
	d.trimTerminalLocked()
	active := 0
	for _, job := range d.jobs {
		if !terminalDeliveryState(job.State) {
			active++
		}
	}
	queueLimit := request.QueueLimit
	if queueLimit > d.maxJobs {
		queueLimit = d.maxJobs
	}
	if active >= queueLimit || len(d.jobs) >= d.maxJobs {
		return Result{}, errors.New("alarm delivery queue is full")
	}
	now := d.clock().UTC()
	d.sequence++
	deliveryID := fmt.Sprintf("alarm_delivery_%s_%06d", now.Format("20060102"), d.sequence)
	job := deliveryJob{
		DeliveryID: deliveryID, State: "queued", Request: request,
		AcceptedAt: now, UpdatedAt: now, NextAttempt: now,
	}
	d.jobs[deliveryID] = job
	if err := d.persistLocked(); err != nil {
		delete(d.jobs, deliveryID)
		d.sequence--
		return Result{}, err
	}
	d.notify()
	return resultFromJob(job), nil
}

func (d *Dispatcher) Status(ctx context.Context, deliveryID string) (Result, error) {
	if err := validateDeliveryID(deliveryID); err != nil {
		return Result{}, err
	}
	if err := ctx.Err(); err != nil {
		return Result{}, err
	}
	d.mu.Lock()
	defer d.mu.Unlock()
	job, ok := d.jobs[deliveryID]
	if !ok {
		return Result{}, errors.New("alarm delivery not found")
	}
	return resultFromJob(job), nil
}

func (d *Dispatcher) Close() {
	d.closeOnce.Do(func() {
		d.cancel()
		d.wg.Wait()
	})
}

func (d *Dispatcher) worker() {
	defer d.wg.Done()
	ticker := time.NewTicker(d.pollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-d.ctx.Done():
			return
		case <-d.wake:
			d.processDue()
		case <-ticker.C:
			d.processDue()
		}
	}
}

func (d *Dispatcher) processDue() {
	for {
		d.mu.Lock()
		now := d.clock().UTC()
		var selected deliveryJob
		found := false
		for _, job := range d.jobs {
			if (job.State != "queued" && job.State != "retry_pending") || job.NextAttempt.After(now) {
				continue
			}
			if !found || job.NextAttempt.Before(selected.NextAttempt) || (job.NextAttempt.Equal(selected.NextAttempt) && job.DeliveryID < selected.DeliveryID) {
				selected = job
				found = true
			}
		}
		if !found {
			d.mu.Unlock()
			return
		}
		selected.State = "running"
		selected.UpdatedAt = now
		d.jobs[selected.DeliveryID] = selected
		if err := d.persistLocked(); err != nil {
			selected.State = "queued"
			d.jobs[selected.DeliveryID] = selected
			d.mu.Unlock()
			return
		}
		d.mu.Unlock()

		attemptCtx, cancel := context.WithTimeout(d.ctx, 10*time.Second)
		err := d.send(attemptCtx, selected)
		cancel()

		d.mu.Lock()
		job, ok := d.jobs[selected.DeliveryID]
		if !ok {
			d.mu.Unlock()
			continue
		}
		now = d.clock().UTC()
		job.Attempts++
		job.UpdatedAt = now
		if err == nil {
			job.State = "delivered"
			job.LastError = ""
			job.CompletedAt = timePointer(now)
		} else {
			job.LastError = boundedDeliveryError(err)
			if job.Attempts <= job.Request.RetryCount {
				job.State = "retry_pending"
				job.NextAttempt = now.Add(time.Duration(job.Request.RetryIntervalSeconds) * time.Second)
			} else {
				job.State = "failed"
				job.CompletedAt = timePointer(now)
			}
		}
		d.jobs[job.DeliveryID] = job
		d.trimTerminalLocked()
		_ = d.persistLocked()
		d.mu.Unlock()
	}
}

func (d *Dispatcher) send(ctx context.Context, job deliveryJob) error {
	for _, channel := range job.Request.Channels {
		transport := d.transports[channel]
		if transport == nil {
			return fmt.Errorf("alarm delivery %s transport is not configured", channel)
		}
		if err := transport.Send(ctx, Attempt{DeliveryID: job.DeliveryID, Channel: channel, Request: job.Request}); err != nil {
			return fmt.Errorf("alarm delivery %s: %w", channel, err)
		}
	}
	return nil
}

func (d *Dispatcher) notify() {
	select {
	case d.wake <- struct{}{}:
	default:
	}
}

func (d *Dispatcher) trimTerminalLocked() {
	if len(d.jobs) < d.maxJobs {
		return
	}
	terminal := make([]deliveryJob, 0)
	for _, job := range d.jobs {
		if terminalDeliveryState(job.State) {
			terminal = append(terminal, job)
		}
	}
	sort.Slice(terminal, func(left, right int) bool {
		if terminal[left].UpdatedAt.Equal(terminal[right].UpdatedAt) {
			return terminal[left].DeliveryID < terminal[right].DeliveryID
		}
		return terminal[left].UpdatedAt.Before(terminal[right].UpdatedAt)
	})
	for _, job := range terminal {
		if len(d.jobs) < d.maxJobs {
			return
		}
		delete(d.jobs, job.DeliveryID)
	}
}

func (d *Dispatcher) load() error {
	pathInfo, err := os.Lstat(d.statePath)
	if err != nil {
		if os.IsNotExist(err) {
			return d.persistLocked()
		}
		return fmt.Errorf("inspect alarm delivery state: %w", err)
	}
	if pathInfo.Mode()&os.ModeSymlink != 0 || !pathInfo.Mode().IsRegular() {
		return errors.New("alarm delivery state must be a bounded regular file")
	}
	file, err := os.Open(d.statePath)
	if err != nil {
		return fmt.Errorf("read alarm delivery state: %w", err)
	}
	info, err := file.Stat()
	if err != nil {
		_ = file.Close()
		return fmt.Errorf("inspect alarm delivery state: %w", err)
	}
	if !info.Mode().IsRegular() || info.Size() < 1 || info.Size() > maxStateBytes {
		_ = file.Close()
		return errors.New("alarm delivery state must be a bounded regular file")
	}
	content, err := io.ReadAll(io.LimitReader(file, maxStateBytes+1))
	if err != nil {
		_ = file.Close()
		return fmt.Errorf("read alarm delivery state: %w", err)
	}
	if err := file.Close(); err != nil {
		return fmt.Errorf("close alarm delivery state: %w", err)
	}
	if len(content) > maxStateBytes {
		return errors.New("alarm delivery state exceeds the configured limit")
	}
	var snapshot dispatcherSnapshot
	decoder := json.NewDecoder(strings.NewReader(string(content)))
	decoder.DisallowUnknownFields()
	if err := decoder.Decode(&snapshot); err != nil {
		return fmt.Errorf("decode alarm delivery state: %w", err)
	}
	var extra any
	if err := decoder.Decode(&extra); !errors.Is(err, io.EOF) {
		return errors.New("alarm delivery state must contain one JSON value")
	}
	if snapshot.FormatVersion != 1 || len(snapshot.Jobs) > d.maxJobs {
		return errors.New("alarm delivery state is invalid or exceeds the configured limit")
	}
	now := d.clock().UTC()
	for deliveryID, job := range snapshot.Jobs {
		if deliveryID != job.DeliveryID || validateDeliveryID(deliveryID) != nil {
			return errors.New("alarm delivery state contains an invalid delivery id")
		}
		normalized, err := NormalizeRequest(job.Request)
		if err != nil || job.AcceptedAt.IsZero() || job.UpdatedAt.IsZero() || job.Attempts < 0 || job.Attempts > normalized.RetryCount+1 {
			return fmt.Errorf("alarm delivery state job %s is invalid", deliveryID)
		}
		job.Request = normalized
		switch job.State {
		case "queued", "retry_pending", "delivered", "failed":
		case "running":
			job.State = "queued"
			job.NextAttempt = now
		default:
			return fmt.Errorf("alarm delivery state job %s has invalid state", deliveryID)
		}
		d.jobs[deliveryID] = job
		if sequence := idSequence(deliveryID); sequence > d.sequence {
			d.sequence = sequence
		}
	}
	if snapshot.Sequence > d.sequence {
		d.sequence = snapshot.Sequence
	}
	return d.persistLocked()
}

func (d *Dispatcher) persistLocked() error {
	snapshot := dispatcherSnapshot{FormatVersion: 1, Sequence: d.sequence, Jobs: make(map[string]deliveryJob, len(d.jobs))}
	for deliveryID, job := range d.jobs {
		snapshot.Jobs[deliveryID] = job
	}
	content, err := json.MarshalIndent(snapshot, "", "  ")
	if err != nil {
		return err
	}
	if err := atomicfile.Write(d.statePath, append(content, '\n'), 0o600); err != nil {
		return fmt.Errorf("persist alarm delivery state: %w", err)
	}
	return nil
}

func ensureStateDirectory(directory string) error {
	if info, err := os.Lstat(directory); err == nil {
		if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
			return errors.New("alarm delivery state directory must be a real directory")
		}
	} else if os.IsNotExist(err) {
		if err := os.MkdirAll(directory, 0o700); err != nil {
			return err
		}
	} else {
		return err
	}
	return os.Chmod(directory, 0o700)
}

func validateDeliveryID(value string) error {
	if len(value) != len("alarm_delivery_20060102_000000") || !strings.HasPrefix(value, "alarm_delivery_") {
		return errors.New("alarm delivery id is invalid")
	}
	dateStart := len("alarm_delivery_")
	if _, err := time.Parse("20060102", value[dateStart:dateStart+8]); err != nil || value[dateStart+8] != '_' {
		return errors.New("alarm delivery id is invalid")
	}
	for _, char := range value[len(value)-6:] {
		if char < '0' || char > '9' {
			return errors.New("alarm delivery id is invalid")
		}
	}
	return nil
}

func terminalDeliveryState(state string) bool { return state == "delivered" || state == "failed" }

func resultFromJob(job deliveryJob) Result {
	return Result{
		DeliveryID: job.DeliveryID, State: job.State, Attempts: job.Attempts,
		LastError: job.LastError, AcceptedAt: job.AcceptedAt, UpdatedAt: job.UpdatedAt, CompletedAt: job.CompletedAt,
	}
}

func boundedDeliveryError(err error) string {
	message := strings.TrimSpace(err.Error())
	message = strings.ReplaceAll(message, "\r", " ")
	message = strings.ReplaceAll(message, "\n", " ")
	runes := []rune(message)
	if len(runes) > 256 {
		message = string(runes[:256])
	}
	if message == "" {
		return "unknown delivery failure"
	}
	return message
}

func timePointer(value time.Time) *time.Time { return &value }

func idSequence(value string) int {
	var sequence int
	_, _ = fmt.Sscanf(value[len(value)-6:], "%d", &sequence)
	return sequence
}