weatherreporter/internal/app/batch_notification.go

package app

import (
	"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/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
}

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

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 := config.RenderDistributorReportPaths(cfg.Notify.Distributor.ReportPathTemplates, values)
		if err != nil {
			return batchNotificationRequest{}, fmt.Errorf("batch notification report %q run %q source path %q: %w", item.ReportID, item.RunID, item.ReportPath, 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 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
}