weatherreporter/internal/app/batch_notification.go

package app

import (
	"context"
	"fmt"
	"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/state"
	"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)
}

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

func notifyBatch(ctx context.Context, cfg config.Config, batch BatchKind, runID string, startedAt time.Time, result *BatchResult, planned []plannedBatchReport, store state.Store, notifier Notifier) (*BatchNotificationResult, error) {
	if !cfg.Notify.Distributor.Enabled {
		return nil, nil
	}
	if !cfg.Notify.Distributor.Batch.Enabled {
		return nil, nil
	}
	if result == nil {
		return nil, fmt.Errorf("batch result is required")
	}
	if result.Failed > 0 {
		return &BatchNotificationResult{
			Status: "skipped",
			Reason: "one or more reports failed",
		}, nil
	}

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

	batchNotifier, err := resolveBatchNotifier(cfg, notifier)
	if err != nil {
		path, saveErr := saveBatchNotificationArtifact(ctx, store, cfg, batch, runID, startedAt, req, nil, err)
		if saveErr != nil {
			return nil, saveErr
		}
		return failedBatchNotificationResult(req, path, err), err
	}

	notification, notifyErr := batchNotifier.NotifyBatch(ctx, req)
	wrappedErr := notifyErr
	if notifyErr != nil {
		wrappedErr = fmt.Errorf("notify batch %q run %q bundle %q: %w", batch, runID, req.BundleID, notifyErr)
	}
	path, saveErr := saveBatchNotificationArtifact(ctx, store, cfg, batch, runID, startedAt, req, notification, wrappedErr)
	if saveErr != nil {
		return nil, saveErr
	}

	batchResult := batchNotificationResult(req, notification, path)
	if wrappedErr != nil {
		batchResult.Status = "failed"
		batchResult.Error = wrappedErr.Error()
		return batchResult, wrappedErr
	}
	return batchResult, nil
}

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.ReportPath == "" {
			return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q is missing managed report path", item.ReportID, item.RunID)
		}

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

		included := BatchNotificationReport{
			ReportID:    item.ReportID,
			RunID:       item.RunID,
			SourcePath:  item.ReportPath,
			BundlePaths: append([]string(nil), bundlePaths...),
		}
		for _, bundlePath := range bundlePaths {
			file := batchNotificationFile{
				ReportID:   item.ReportID,
				RunID:      item.RunID,
				SourcePath: item.ReportPath,
				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.ReportPath, 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, path string) *BatchNotificationResult {
	notification := &BatchNotificationResult{
		Status:          "unknown",
		PipelineID:      req.PipelineID,
		BundleID:        req.BundleID,
		IdempotencyKey:  req.IdempotencyKey,
		Path:            path,
		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 = result.Error
		}
	}
	if notification.Status == "" {
		notification.Status = "unknown"
	}
	return notification
}

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

func saveBatchNotificationArtifact(ctx context.Context, store state.Store, cfg config.Config, batch BatchKind, runID string, startedAt time.Time, req batchNotificationRequest, result *NotificationResult, notifyErr error) (string, error) {
	if store == nil {
		return "", fmt.Errorf("state store is required")
	}
	location, err := timeutil.LoadLocation(cfg.WeatherAPI.Timezone)
	if err != nil {
		return "", fmt.Errorf("load batch notification timezone: %w", err)
	}
	artifact := state.BatchDistributorNotificationArtifact{
		SchemaVersion:  state.BatchDistributorNotificationSchemaVersion,
		Batch:          string(batch),
		BatchRunID:     runID,
		AttemptedAt:    time.Now(),
		Endpoint:       cfg.Notify.Distributor.Endpoint,
		PipelineID:     req.PipelineID,
		BundleID:       req.BundleID,
		IdempotencyKey: req.IdempotencyKey,
		BundleCreated:  req.CreatedAt,
		Reports:        batchNotificationReportArtifacts(req.IncludedReports),
		Status:         "attempted",
	}
	if result != nil {
		artifact.Status = result.Status
		artifact.Upload = &state.DistributorUploadResult{
			RunID:  result.RunID,
			Status: result.UploadStatus,
		}
		if result.PipelineID != "" || !result.AcceptedAt.IsZero() || result.StartedAt != nil || result.FinishedAt != nil || len(result.Report) > 0 || result.Error != "" {
			artifact.RunStatus = &state.DistributorRunStatus{
				RunID:      result.RunID,
				PipelineID: result.PipelineID,
				Status:     result.Status,
				AcceptedAt: result.AcceptedAt,
				StartedAt:  result.StartedAt,
				FinishedAt: result.FinishedAt,
				Report:     append([]byte(nil), result.Report...),
				Error:      result.Error,
			}
		}
		artifact.StatusError = result.StatusError
	}
	if notifyErr != nil {
		artifact.Status = "failed"
		artifact.Error = notifyErr.Error()
	}
	if artifact.Status == "" {
		artifact.Status = "unknown"
	}
	return store.SaveBatchDistributorNotification(ctx, state.BatchDistributorNotificationRef{
		Batch:      string(batch),
		BatchRunID: runID,
		StartedAt:  startedAt,
		Location:   location,
	}, artifact)
}

func batchNotificationReportArtifacts(reports []BatchNotificationReport) []state.BatchDistributorNotificationReportArtifact {
	if len(reports) == 0 {
		return nil
	}
	artifacts := make([]state.BatchDistributorNotificationReportArtifact, 0, len(reports))
	for _, item := range reports {
		artifacts = append(artifacts, state.BatchDistributorNotificationReportArtifact{
			ReportID:    item.ReportID,
			RunID:       item.RunID,
			SourcePath:  item.SourcePath,
			BundlePaths: append([]string(nil), item.BundlePaths...),
		})
	}
	return artifacts
}

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
}