|
| 1 | +package report |
| 2 | + |
| 3 | +// Async report rendering: the bulk attestation faces (CSV / OSCAL SAR / |
| 4 | +// PDF) can be expensive to assemble for a large fleet, so generating an |
| 5 | +// attestation enqueues a render job instead of blocking the request. A |
| 6 | +// worker claims the job, renders each face (warming the report_faces cache |
| 7 | +// and flipping each face's row from 'pending' to 'ready'), then publishes |
| 8 | +// a ReportReady event on the bus - the first producer of the in-app |
| 9 | +// notification bell. |
| 10 | +// |
| 11 | +// Export stays the lazy fallback: a download that arrives before the job |
| 12 | +// runs still renders the face inline (a cache miss), so async rendering is |
| 13 | +// a warm-the-cache-and-notify optimization, never a correctness gate. |
| 14 | +// |
| 15 | +// Spec: api-reports. |
| 16 | + |
| 17 | +import ( |
| 18 | + "context" |
| 19 | + "encoding/json" |
| 20 | + "fmt" |
| 21 | + "log/slog" |
| 22 | + "time" |
| 23 | + |
| 24 | + "github.com/google/uuid" |
| 25 | + |
| 26 | + "github.com/Hanalyx/openwatch/internal/eventbus" |
| 27 | + "github.com/Hanalyx/openwatch/internal/queue" |
| 28 | +) |
| 29 | + |
| 30 | +// RenderJobType is the job_type for an async report-face render. |
| 31 | +const RenderJobType = "report.render" |
| 32 | + |
| 33 | +// RenderPayload is the job payload: which snapshot to render faces for. |
| 34 | +type RenderPayload struct { |
| 35 | + SnapshotID uuid.UUID `json:"snapshot_id"` |
| 36 | +} |
| 37 | + |
| 38 | +// attestationFaces are the bulk faces rendered asynchronously for an |
| 39 | +// attestation report, in render order. |
| 40 | +var attestationFaces = []string{FaceCSV, FaceOSCALSAR, FacePDF} |
| 41 | + |
| 42 | +// markFacesPending inserts a 'pending' report_faces row for each face that |
| 43 | +// does not already exist, so the lifecycle status is genuine (the render |
| 44 | +// worker flips them to 'ready'). ON CONFLICT DO NOTHING leaves an already |
| 45 | +// rendered ('ready') face untouched. |
| 46 | +func (s *Service) markFacesPending(ctx context.Context, snapshotID uuid.UUID, faces []string) error { |
| 47 | + for _, face := range faces { |
| 48 | + mediaType := "application/octet-stream" |
| 49 | + switch face { |
| 50 | + case FaceCSV: |
| 51 | + mediaType = "text/csv" |
| 52 | + case FaceOSCALSAR, FaceJSON: |
| 53 | + mediaType = "application/json" |
| 54 | + case FacePDF: |
| 55 | + mediaType = "application/pdf" |
| 56 | + } |
| 57 | + if _, err := s.pool.Exec(ctx, ` |
| 58 | + INSERT INTO report_faces (snapshot_id, face, media_type, size_bytes, status) |
| 59 | + VALUES ($1, $2, $3, 0, 'pending') |
| 60 | + ON CONFLICT (snapshot_id, face) DO NOTHING`, |
| 61 | + snapshotID, face, mediaType); err != nil { |
| 62 | + return fmt.Errorf("report: mark face pending: %w", err) |
| 63 | + } |
| 64 | + } |
| 65 | + return nil |
| 66 | +} |
| 67 | + |
| 68 | +// enqueueRender marks the attestation's bulk faces pending and enqueues a |
| 69 | +// render job. Best-effort: a failure to enqueue is logged but does not fail |
| 70 | +// Generate, because Export still renders each face lazily on first |
| 71 | +// download (the async path is an optimization, not a correctness gate). |
| 72 | +func (s *Service) enqueueRender(ctx context.Context, snapshotID uuid.UUID) { |
| 73 | + if err := s.markFacesPending(ctx, snapshotID, attestationFaces); err != nil { |
| 74 | + slog.WarnContext(ctx, "report: mark faces pending failed", |
| 75 | + slog.String("snapshot_id", snapshotID.String()), slog.String("error", err.Error())) |
| 76 | + return |
| 77 | + } |
| 78 | + if _, err := queue.Enqueue(ctx, s.pool, RenderJobType, RenderPayload{SnapshotID: snapshotID}); err != nil { |
| 79 | + slog.WarnContext(ctx, "report: enqueue render job failed", |
| 80 | + slog.String("snapshot_id", snapshotID.String()), slog.String("error", err.Error())) |
| 81 | + } |
| 82 | +} |
| 83 | + |
| 84 | +// RenderProcessor renders a report's faces for a claimed report.render job |
| 85 | +// and publishes ReportReady. It is registered on the in-process worker via |
| 86 | +// WithReportProcessor; its ProcessJob signature matches the worker's other |
| 87 | +// processors. |
| 88 | +type RenderProcessor struct { |
| 89 | + svc *Service |
| 90 | + bus *eventbus.Bus |
| 91 | +} |
| 92 | + |
| 93 | +// NewRenderProcessor builds the processor over a report Service (for |
| 94 | +// Export) and an event bus (to publish ReportReady). A nil bus renders the |
| 95 | +// faces but publishes nothing. |
| 96 | +func NewRenderProcessor(svc *Service, bus *eventbus.Bus) *RenderProcessor { |
| 97 | + return &RenderProcessor{svc: svc, bus: bus} |
| 98 | +} |
| 99 | + |
| 100 | +// ProcessJob renders every face that applies to the report's kind (warming |
| 101 | +// the cache and flipping each 'pending' row to 'ready' via Export's upsert) |
| 102 | +// and publishes a ReportReady event. A render error fails the job so it can |
| 103 | +// be retried; faces already rendered are idempotent (deterministic bytes). |
| 104 | +func (p *RenderProcessor) ProcessJob(ctx context.Context, j *queue.Job) { |
| 105 | + var payload RenderPayload |
| 106 | + if err := json.Unmarshal(j.Payload, &payload); err != nil { |
| 107 | + _ = queue.Fail(ctx, p.svc.pool, j.ID, fmt.Sprintf("report.render: payload decode: %v", err)) |
| 108 | + return |
| 109 | + } |
| 110 | + if payload.SnapshotID == uuid.Nil { |
| 111 | + _ = queue.Fail(ctx, p.svc.pool, j.ID, "report.render: payload snapshot_id missing") |
| 112 | + return |
| 113 | + } |
| 114 | + |
| 115 | + rep, err := p.svc.Get(ctx, payload.SnapshotID) |
| 116 | + if err != nil { |
| 117 | + // An unknown snapshot (e.g. deleted before the job ran) is terminal, |
| 118 | + // not retryable. |
| 119 | + _ = queue.Fail(ctx, p.svc.pool, j.ID, fmt.Sprintf("report.render: load snapshot: %v", err)) |
| 120 | + return |
| 121 | + } |
| 122 | + |
| 123 | + faces := facesForKind(rep.Kind) |
| 124 | + rendered := make([]string, 0, len(faces)) |
| 125 | + for _, face := range faces { |
| 126 | + if _, _, err := p.svc.Export(ctx, payload.SnapshotID, face); err != nil { |
| 127 | + // Mark the face failed so the lifecycle reflects reality, then |
| 128 | + // fail the job for retry. |
| 129 | + _, _ = p.svc.pool.Exec(ctx, |
| 130 | + `UPDATE report_faces SET status = 'failed' WHERE snapshot_id = $1 AND face = $2 AND status = 'pending'`, |
| 131 | + payload.SnapshotID, face) |
| 132 | + _ = queue.Fail(ctx, p.svc.pool, j.ID, fmt.Sprintf("report.render: face %s: %v", face, err)) |
| 133 | + return |
| 134 | + } |
| 135 | + rendered = append(rendered, face) |
| 136 | + } |
| 137 | + |
| 138 | + if p.bus != nil { |
| 139 | + p.bus.Publish(ctx, eventbus.ReportReady{ |
| 140 | + SnapshotID: rep.ID, |
| 141 | + ReportKind: string(rep.Kind), |
| 142 | + Faces: rendered, |
| 143 | + GeneratedBy: rep.GeneratedBy, |
| 144 | + OccurredAt: time.Now().UTC(), |
| 145 | + }) |
| 146 | + } |
| 147 | + |
| 148 | + if err := queue.Complete(ctx, p.svc.pool, j.ID); err != nil { |
| 149 | + slog.WarnContext(ctx, "report.render: complete failed", |
| 150 | + slog.String("job_id", j.ID.String()), slog.String("error", err.Error())) |
| 151 | + } |
| 152 | +} |
| 153 | + |
| 154 | +// facesForKind lists the faces an async render produces for a report kind: |
| 155 | +// the bulk faces for an attestation; just the (cheap) PDF for an executive |
| 156 | +// so the executive path can also notify if ever enqueued. The JSON face is |
| 157 | +// always available lazily and is not pre-rendered. |
| 158 | +func facesForKind(kind Kind) []string { |
| 159 | + switch kind { |
| 160 | + case KindAttestation: |
| 161 | + return attestationFaces |
| 162 | + case KindExecutive: |
| 163 | + return []string{FacePDF} |
| 164 | + default: |
| 165 | + return nil |
| 166 | + } |
| 167 | +} |
0 commit comments