weatherreporter/internal/app/batch_notification.go

package app

import (
	"context"
	"fmt"
	"path/filepath"
	"time"

	distributoradapter "gitea.maximumdirect.net/eric/weatherreporter/internal/adapters/distributor"
	"gitea.maximumdirect.net/eric/weatherreporter/internal/config"
	"gitea.maximumdirect.net/eric/weatherreporter/internal/report"
	"gitea.maximumdirect.net/eric/weatherreporter/internal/timeutil"
)

const runIDTimestampLayout = "20060102T150405.000000000Z"

type batchNotificationIdentity struct {
	PipelineID     string
	BundleID       string
	IdempotencyKey string
}

type batchNotificationRequest struct {
	Batch           BatchKind
	RunID           string
	PipelineID      string
	BundleID        string
	IdempotencyKey  string
	Files           []batchNotificationFile
	IncludedReports []BatchNotificationReport
	CreatedAt       time.Time
}

type batchNotificationFile struct {
	ReportID   report.ID
	RunID      string
	SourcePath string
	BundlePath string
}

type batchNotifier interface {
	NotifyBatch(context.Context, batchNotificationRequest) (*NotificationResult, error)
}

type batchNotificationInput struct {
	ctx          context.Context
	cancellation error
	cfg          config.Config
	batch        BatchKind
	runID        string
	startedAt    time.Time
	result       *BatchResult
	planned      []plannedBatchReport
	notifier     Notifier
}

func batchRunID(startedAt time.Time, batch BatchKind) string {
	return startedAt.UTC().Format(runIDTimestampLayout) + "_" + string(batch)
}

func notifyBatch(input batchNotificationInput) *BatchNotificationResult {
	if !input.cfg.Notify.Distributor.Enabled {
		return nil
	}
	if !input.cfg.Notify.Distributor.Batch.Enabled {
		return nil
	}
	if input.result == nil {
		return failedBatchNotificationResult(batchNotificationRequest{}, fmt.Errorf("batch result is required"))
	}
	if input.cancellation != nil || batchContextCancellationCause(input.ctx) != nil || input.result.Canceled > 0 {
		return &BatchNotificationResult{
			Status: "skipped",
			Reason: "batch canceled",
		}
	}
	if input.result.Failed > 0 {
		return &BatchNotificationResult{
			Status: "skipped",
			Reason: "one or more reports failed",
		}
	}

	req, err := buildBatchNotificationRequest(input.cfg, input.batch, input.runID, input.startedAt, input.result.Reports, input.planned)
	if err != nil {
		return failedBatchNotificationResult(batchNotificationRequest{}, err)
	}

	batchNotifier, err := resolveBatchNotifier(input.cfg, input.notifier)
	if err != nil {
		return failedBatchNotificationResult(req, err)
	}

	notification, notifyErr := batchNotifier.NotifyBatch(input.ctx, req)
	wrappedErr := notifyErr
	if notifyErr != nil {
		wrappedErr = fmt.Errorf("notify batch %q run %q bundle %q: %w", input.batch, input.runID, req.BundleID, notifyErr)
	}
	batchResult := batchNotificationResult(req, notification)
	if wrappedErr != nil {
		batchResult.Status = "failed"
		batchResult.Error = safeDistributorNotificationFailure(wrappedErr)
		return batchResult
	}
	return batchResult
}

func resolveBatchNotifier(cfg config.Config, notifier Notifier) (batchNotifier, error) {
	if notifier != nil {
		if batchNotifier, ok := notifier.(batchNotifier); ok {
			return batchNotifier, nil
		}
		return nil, fmt.Errorf("batch distributor notifier is required")
	}
	return distributorNotifier{
		client: distributoradapter.New(cfg.Notify.Distributor),
	}, nil
}

func buildBatchNotificationRequest(cfg config.Config, batch BatchKind, runID string, startedAt time.Time, reports []BatchReportResult, planned []plannedBatchReport) (batchNotificationRequest, error) {
	if len(reports) == 0 {
		return batchNotificationRequest{}, fmt.Errorf("batch notification requires at least one report")
	}

	identity, err := renderBatchNotificationIdentity(cfg, batch, runID, startedAt)
	if err != nil {
		return batchNotificationRequest{}, err
	}
	if identity.PipelineID == "" {
		return batchNotificationRequest{}, fmt.Errorf("batch notification pipeline id is required")
	}
	if identity.BundleID == "" {
		return batchNotificationRequest{}, fmt.Errorf("batch notification bundle id is required")
	}
	if identity.IdempotencyKey == "" {
		return batchNotificationRequest{}, fmt.Errorf("batch notification idempotency key is required for bundle %q", identity.BundleID)
	}

	plannedByRunID, err := plannedReportsByRunID(planned)
	if err != nil {
		return batchNotificationRequest{}, err
	}

	req := batchNotificationRequest{
		Batch:          batch,
		RunID:          runID,
		PipelineID:     identity.PipelineID,
		BundleID:       identity.BundleID,
		IdempotencyKey: identity.IdempotencyKey,
		CreatedAt:      startedAt,
	}
	seenBundlePaths := map[string]batchNotificationFile{}
	for _, item := range reports {
		plannedReport, ok := plannedByRunID[item.RunID]
		if !ok {
			return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q has no matching planned report", item.ReportID, item.RunID)
		}
		if item.ReportID != plannedReport.Resolved.Definition.ID {
			return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q does not match planned report %q", item.ReportID, item.RunID, plannedReport.Resolved.Definition.ID)
		}
		if item.OutputPath == "" {
			return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q is missing output path", item.ReportID, item.RunID)
		}

		values, err := distributorTemplateValuesForReport(cfg, plannedReport.Resolved, item.RunID, filepath.Base(item.OutputPath))
		if err != nil {
			return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q source path %q: %w", item.ReportID, item.RunID, item.OutputPath, err)
		}
		bundlePaths, err := renderDistributorReportBundlePaths(cfg, plannedReport.Resolved, item.RunID, item.OutputPath, values)
		if err != nil {
			return batchNotificationRequest{}, err
		}

		included := BatchNotificationReport{
			ReportID:    item.ReportID,
			RunID:       item.RunID,
			SourcePath:  item.OutputPath,
			BundlePaths: append([]string(nil), bundlePaths...),
		}
		for _, bundlePath := range bundlePaths {
			file := batchNotificationFile{
				ReportID:   item.ReportID,
				RunID:      item.RunID,
				SourcePath: item.OutputPath,
				BundlePath: bundlePath,
			}
			if previous, ok := seenBundlePaths[bundlePath]; ok {
				return batchNotificationRequest{}, fmt.Errorf("batch notification duplicate bundle path %q for report %q run %q source path %q; already used by report %q run %q source path %q", bundlePath, item.ReportID, item.RunID, item.OutputPath, previous.ReportID, previous.RunID, previous.SourcePath)
			}
			seenBundlePaths[bundlePath] = file
			req.Files = append(req.Files, file)
		}
		req.IncludedReports = append(req.IncludedReports, included)
	}
	if len(req.Files) == 0 {
		return batchNotificationRequest{}, fmt.Errorf("batch notification requires at least one file mapping")
	}
	return req, nil
}

func plannedReportsByRunID(planned []plannedBatchReport) (map[string]plannedBatchReport, error) {
	byRunID := make(map[string]plannedBatchReport, len(planned))
	for _, item := range planned {
		runID := item.Resolved.Metadata().RunID
		if runID == "" {
			return nil, fmt.Errorf("planned report %q has empty run id", item.Resolved.Definition.ID)
		}
		if previous, ok := byRunID[runID]; ok {
			return nil, fmt.Errorf("planned reports %q and %q share run id %q", previous.Resolved.Definition.ID, item.Resolved.Definition.ID, runID)
		}
		byRunID[runID] = item
	}
	return byRunID, nil
}

func batchDistributorUploadRequest(req batchNotificationRequest) distributoradapter.UploadRequest {
	files := make([]distributoradapter.UploadFile, 0, len(req.Files))
	for _, file := range req.Files {
		files = append(files, distributoradapter.UploadFile{
			SourcePath: file.SourcePath,
			BundlePath: file.BundlePath,
		})
	}
	return distributoradapter.UploadRequest{
		PipelineID:     req.PipelineID,
		BundleID:       req.BundleID,
		IdempotencyKey: req.IdempotencyKey,
		Files:          files,
		CreatedAt:      req.CreatedAt,
	}
}

func batchNotificationResult(req batchNotificationRequest, result *NotificationResult) *BatchNotificationResult {
	notification := &BatchNotificationResult{
		Status:          "unknown",
		PipelineID:      req.PipelineID,
		BundleID:        req.BundleID,
		IdempotencyKey:  req.IdempotencyKey,
		IncludedReports: append([]BatchNotificationReport(nil), req.IncludedReports...),
	}
	if result != nil {
		notification.Status = result.Status
		notification.RunID = result.RunID
		if result.PipelineID != "" {
			notification.PipelineID = result.PipelineID
		}
		if result.BundleID != "" {
			notification.BundleID = result.BundleID
		}
		if result.IdempotencyKey != "" {
			notification.IdempotencyKey = result.IdempotencyKey
		}
		if result.Error != "" {
			notification.Error = safeDistributorRunError(result.Error)
		}
	}
	if notification.Status == "" {
		notification.Status = "unknown"
	}
	return notification
}

func failedBatchNotificationResult(req batchNotificationRequest, err error) *BatchNotificationResult {
	notification := batchNotificationResult(req, nil)
	notification.Status = "failed"
	if err != nil {
		notification.Error = safeDistributorNotificationFailure(err)
	}
	return notification
}

func safeDistributorNotificationFailure(err error) string {
	if err == nil {
		return ""
	}
	return "distributor notification failed"
}

func renderBatchNotificationIdentity(cfg config.Config, batch BatchKind, runID string, startedAt time.Time) (batchNotificationIdentity, error) {
	values, err := batchNotificationTemplateValues(cfg, batch, runID, startedAt)
	if err != nil {
		return batchNotificationIdentity{}, err
	}

	bundleID, err := config.RenderDistributorBatchBundleID(cfg.Notify.Distributor.Batch.BundleIDTemplate, values)
	if err != nil {
		return batchNotificationIdentity{}, err
	}
	values.BundleID = bundleID

	pipelineID, err := config.RenderDistributorBatchPipelineID(cfg.Notify.Distributor.Batch.PipelineIDTemplate, values)
	if err != nil {
		return batchNotificationIdentity{}, err
	}
	idempotencyKey, err := config.RenderDistributorBatchIdempotencyKey(cfg.Notify.Distributor.Batch.IdempotencyKeyTemplate, values)
	if err != nil {
		return batchNotificationIdentity{}, err
	}

	return batchNotificationIdentity{
		PipelineID:     pipelineID,
		BundleID:       bundleID,
		IdempotencyKey: idempotencyKey,
	}, nil
}

func batchNotificationTemplateValues(cfg config.Config, batch BatchKind, runID string, startedAt time.Time) (config.DistributorBatchTemplateValues, error) {
	location, err := timeutil.LoadLocation(cfg.WeatherAPI.Timezone)
	if err != nil {
		return config.DistributorBatchTemplateValues{}, fmt.Errorf("load batch notification timezone: %w", err)
	}
	return config.DistributorBatchTemplateValues{
		LocationID:       cfg.Location.ID,
		Batch:            string(batch),
		BatchRunID:       runID,
		BatchStartedDate: startedAt.In(location).Format(timeutil.DateLayout),
	}, nil
}