| 96 | } |
| 97 | |
| 98 | func (s SpoolService) Enqueue(req SpoolEnqueueRequest) (SpoolBatch, error) { |
| 99 | if s.DB == nil { |
| 100 | return SpoolBatch{}, fmt.Errorf("database is required") |
| 101 | } |
| 102 | if err := s.applyBackpressure(req.MaxQueued, req.DropPolicy); err != nil { |
| 103 | return SpoolBatch{}, err |
| 104 | } |
| 105 | if req.Format == "" { |
| 106 | req.Format = "falco" |
| 107 | } |
| 108 | if req.Format != "falco" { |
| 109 | return SpoolBatch{}, fmt.Errorf("unsupported spool format %q", req.Format) |
| 110 | } |
| 111 | if req.SourcePath == "" { |
| 112 | return SpoolBatch{}, fmt.Errorf("source_path is required") |
| 113 | } |
| 114 | if err := os.MkdirAll(s.Paths.Spool, 0o755); err != nil { |
| 115 | return SpoolBatch{}, err |
| 116 | } |
| 117 | source, err := os.Open(req.SourcePath) |
| 118 | if err != nil { |
| 119 | return SpoolBatch{}, err |
| 120 | } |
| 121 | defer source.Close() |
| 122 | id := ids.New("spool") |
| 123 | spoolPath := filepath.Join(s.Paths.Spool, id+".jsonl") |
| 124 | target, err := os.OpenFile(spoolPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644) |
| 125 | if err != nil { |
| 126 | return SpoolBatch{}, err |
| 127 | } |
| 128 | hasher := sha256.New() |
| 129 | written, copyErr := io.Copy(io.MultiWriter(target, hasher), source) |
| 130 | closeErr := target.Close() |
| 131 | if copyErr != nil { |
| 132 | _ = os.Remove(spoolPath) |
| 133 | return SpoolBatch{}, copyErr |
| 134 | } |
| 135 | if closeErr != nil { |
| 136 | _ = os.Remove(spoolPath) |
| 137 | return SpoolBatch{}, closeErr |
| 138 | } |
| 139 | now := time.Now().UTC().Format(time.RFC3339Nano) |
| 140 | policyEnabled := 0 |
| 141 | if req.PolicyEnabled { |
| 142 | policyEnabled = 1 |
| 143 | } |
| 144 | hash := hex.EncodeToString(hasher.Sum(nil)) |
| 145 | _, err = s.DB.Exec(`INSERT INTO telemetry_spool_batches |
| 146 | (id, run_id, format, source_path, spool_path, file_sha256, size_bytes, status, policy_enabled, created_at, updated_at) |
| 147 | VALUES (?, ?, ?, ?, ?, ?, ?, 'queued', ?, ?, ?)`, |
| 148 | id, req.RunID, req.Format, req.SourcePath, spoolPath, hash, written, policyEnabled, now, now) |
| 149 | if err != nil { |
| 150 | _ = os.Remove(spoolPath) |
| 151 | return SpoolBatch{}, err |
| 152 | } |
| 153 | return SpoolBatch{ |
| 154 | ID: id, |
| 155 | RunID: req.RunID, |