mrcharles
05ec1a4142
Signature verification (gitea hex HMAC-SHA256 over the raw body via constant-time hmac.Equal; redmine/discourse constant-time shared-secret headers) with one generic ErrUnverified so rejects give attackers no oracle. Tolerant normalization of the known Redmine/Discourse/Gitea payload variants into one Event record (canonical subject ids, actor, title, repo, sha256 payload digest) plus the DESIGN action mapping (dispatch_turn / respond_turn / pipeline_step / ignore). Store: JSONL under state dir, 0600, dedup keyed on provider event id (delivery header, payload-digest fallback), index rebuilt at startup so replays across restarts still dedup; torn tail lines skipped, not fatal.
102 lines
2.6 KiB
Go
102 lines
2.6 KiB
Go
package events
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"fmt"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
)
|
|
|
|
// Store is the append-only event log: one Event per JSON line under
|
|
// stateDir/events.jsonl, deduped by provider event id (source:providerID).
|
|
// The dedup index is rebuilt from the file at open, so restarts keep the
|
|
// at-least-once semantics (a torn tail line is skipped, not fatal).
|
|
type Store struct {
|
|
mu sync.Mutex
|
|
path string
|
|
f *os.File
|
|
seen map[string]bool
|
|
}
|
|
|
|
// OpenStore creates/opens the state dir and the event log.
|
|
func OpenStore(stateDir string) (*Store, error) {
|
|
if err := os.MkdirAll(stateDir, 0o700); err != nil {
|
|
return nil, fmt.Errorf("events state dir: %w", err)
|
|
}
|
|
path := filepath.Join(stateDir, "events.jsonl")
|
|
f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("open event log: %w", err)
|
|
}
|
|
s := &Store{path: path, f: f, seen: make(map[string]bool)}
|
|
if err := s.loadIndex(); err != nil {
|
|
f.Close()
|
|
return nil, err
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
// loadIndex populates the dedup set from the existing log.
|
|
func (s *Store) loadIndex() error {
|
|
rf, err := os.Open(s.path)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("read event log: %w", err)
|
|
}
|
|
defer rf.Close()
|
|
sc := bufio.NewScanner(rf)
|
|
sc.Buffer(make([]byte, 0, 64*1024), 1<<20)
|
|
for sc.Scan() {
|
|
var ev Event
|
|
if err := json.Unmarshal(sc.Bytes(), &ev); err != nil {
|
|
continue // torn/malformed tail line; dedup stays conservative
|
|
}
|
|
if ev.ProviderID != "" {
|
|
s.seen[ev.DedupKey()] = true
|
|
}
|
|
}
|
|
return sc.Err()
|
|
}
|
|
|
|
// Append records ev unless its provider event id is already logged.
|
|
// stored=false means duplicate (replay); err means the write failed and the
|
|
// event is NOT deduped (safe to retry).
|
|
func (s *Store) Append(ev Event) (stored bool, err error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
key := ev.DedupKey()
|
|
if s.seen[key] {
|
|
return false, nil
|
|
}
|
|
line, err := json.Marshal(ev)
|
|
if err != nil {
|
|
return false, fmt.Errorf("encode event: %w", err)
|
|
}
|
|
if _, err := s.f.Write(append(line, '\n')); err != nil {
|
|
return false, fmt.Errorf("append event log: %w", err)
|
|
}
|
|
s.seen[key] = true
|
|
return true, nil
|
|
}
|
|
|
|
// Len reports how many distinct events the index holds (ops/tests).
|
|
func (s *Store) Len() int {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return len(s.seen)
|
|
}
|
|
|
|
// Path exposes the log location (ops messages only; contents stay put).
|
|
func (s *Store) Path() string { return s.path }
|
|
|
|
// Close closes the underlying file.
|
|
func (s *Store) Close() error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.f.Close()
|
|
}
|