Stream and batch

// Run feeds events in arrival order through the windows. The watermark is the largest event time
// seen minus Delay. A window fires when the watermark reaches its end. A late event for a fired
// window updates it within Lateness; after that the event is dropped.
func Run(events []Event, cfg Config) (Result, error) {
  if err := cfg.Window.valid(); err != nil {
    return Result{}, err
  }
  in := slices.Clone(events)
  slices.SortStableFunc(in, func(a, b Event) int { return cmp.Or(cmp.Compare(a.Arrive, b.Arrive), cmp.Compare(a.ID, b.ID)) })
  res := Result{Status: map[int]EventStatus{}}
  var open []*state
  watermark := math.MinInt
  maxSeen := math.MinInt

  for _, e := range in {
    var touched []*state
    late := false
    if cfg.Window.Kind == Session {
      seed := assign(cfg.Window, e.At)[0]
      if seed.End+cfg.Lateness > watermark {
        st := mergeSession(&open, seed)
        touched = append(touched, st)
        late = seed.End <= watermark || st.fired
      }
    } else {
      for _, w := range assign(cfg.Window, e.At) {
        if w.End+cfg.Lateness <= watermark {
          continue // past its lateness: this window's state is gone
        }
        st := find(open, w)
        if st == nil {
          st = &state{w: w}
          open = append(open, st)
        }
        st.count++
        touched = append(touched, st)
        late = late || w.End <= watermark
      }
    }
    switch {
    case len(touched) == 0:
      res.Status[e.ID] = Dropped
    case late:
      res.Status[e.ID] = Kept
    default:
      res.Status[e.ID] = OnTime
    }
    // A window that already fired emits its corrected count at once.
    for _, st := range touched {
      if st.fired {
        res.Emissions = append(res.Emissions, Emission{st.w, st.count, e.Arrive, true})
      }
    }
    maxSeen = max(maxSeen, e.At)
    watermark = max(watermark, maxSeen-cfg.Delay)
    res.Marks = append(res.Marks, Mark{e.Arrive, watermark})
    res.Emissions = append(res.Emissions, fire(&open, watermark, cfg.Lateness, e.Arrive)...)
  }
  // The end of the input moves the watermark to the end of time: every window fires.
  res.Emissions = append(res.Emissions, fire(&open, math.MaxInt-cfg.Lateness, cfg.Lateness, -1)...)
  return res, nil
}

// fire emits every window whose end the watermark has reached, once, and forgets windows whose
// lateness has run out.
func fire(open *[]*state, watermark, lateness, at int) []Emission {
  var out []Emission
  slices.SortFunc(*open, func(a, b *state) int { return cmp.Compare(a.w.Start, b.w.Start) })
  for _, st := range *open {
    if !st.fired && st.w.End <= watermark {
      st.fired = true
      out = append(out, Emission{st.w, st.count, at, false})
    }
  }
  *open = slices.DeleteFunc(*open, func(st *state) bool { return st.w.End+lateness <= watermark })
  return out
}

// Restore loads progress after a start or a crash. With auto-commit only the offset survives;
// a checkpoint brings back the offset and the state that matches it.
func (j *Job) Restore(ctx context.Context) error {
  j.next, j.state = 0, map[key]int{}
  var raw []byte
  err := j.DB.QueryRow(ctx, jq["load"], j.Name).Scan(&j.next, &raw)
  if errors.Is(err, pgx.ErrNoRows) {
    return nil
  }
  if err != nil {
    return fmt.Errorf("load progress of %s: %w", j.Name, err)
  }
  if j.Mode == AutoCommit || raw == nil {
    return nil
  }
  var saved []struct {
    K key `json:"k"`
    N int `json:"n"`
  }
  if err := json.Unmarshal(raw, &saved); err != nil {
    return fmt.Errorf("decode state of %s: %w", j.Name, err)
  }
  for _, s := range saved {
    j.state[s.K] = s.N
  }
  return nil
}

// Process handles the click at offset j.Next() and, every Every events, commits or checkpoints.
func (j *Job) Process(ctx context.Context, c Click) error {
  k := key{c.At - c.At%WindowSize, c.Page}
  j.state[k]++
  var err error
  switch j.Mode {
  case AutoCommit, CheckpointAdd:
    _, err = j.DB.Exec(ctx, jq["add"], k.Window, k.Page)
  case CheckpointPut:
    _, err = j.DB.Exec(ctx, jq["put"], k.Window, k.Page, j.state[k])
  case Transactional:
    if j.tx == nil {
      if j.tx, err = j.DB.Begin(ctx); err != nil {
        return fmt.Errorf("begin: %w", err)
      }
    }
    _, err = j.tx.Exec(ctx, jq["put"], k.Window, k.Page, j.state[k])
  default:
    return fmt.Errorf("unknown mode %q", j.Mode)
  }
  if err != nil {
    return fmt.Errorf("write %+v: %w", k, err)
  }
  j.next++
  if j.next%j.Every == 0 {
    return j.Checkpoint(ctx)
  }
  return nil
}

// Checkpoint saves progress. In Transactional mode the sink writes since the last checkpoint
// commit in the same transaction as the offset and the state.
func (j *Job) Checkpoint(ctx context.Context) error {
  var state []byte
  if j.Mode != AutoCommit {
    type entry struct {
      K key `json:"k"`
      N int `json:"n"`
    }
    var all []entry
    for k, n := range j.state {
      all = append(all, entry{k, n})
    }
    var err error
    if state, err = json.Marshal(all); err != nil {
      return fmt.Errorf("encode state: %w", err)
    }
  }
  if j.Mode != Transactional {
    if _, err := j.DB.Exec(ctx, jq["save"], j.Name, j.next, state); err != nil {
      return fmt.Errorf("save progress at %d: %w", j.next, err)
    }
    return nil
  }
  if j.tx == nil {
    return nil
  }
  if _, err := j.tx.Exec(ctx, jq["save"], j.Name, j.next, state); err != nil {
    return fmt.Errorf("save progress at %d: %w", j.next, err)
  }
  err := j.tx.Commit(ctx)
  j.tx = nil
  if err != nil {
    return fmt.Errorf("commit at %d: %w", j.next, err)
  }
  return nil
}

-- Idempotent: writes the count from the job's state. A replay writes the same value again.
INSERT INTO window_counts (window_start, page, clicks) VALUES ($1, $2, $3)
ON CONFLICT (window_start, page) DO UPDATE SET clicks = EXCLUDED.clicks;

-- The checkpoint: the next offset and the state, in one row, so they always match.
INSERT INTO job_progress (job, next_offset, state) VALUES ($1, $2, $3)
ON CONFLICT (job) DO UPDATE SET next_offset = EXCLUDED.next_offset, state = EXCLUDED.state;

SELECT next_offset, state FROM job_progress WHERE job = $1;

-- The output: clicks per page per 10-second window, read by a dashboard.
CREATE TABLE window_counts (
  window_start int  NOT NULL,
  page         text NOT NULL,
  clicks       int  NOT NULL,
  PRIMARY KEY (window_start, page)
);

-- Where the job is in the log. An auto-committing consumer stores only the offset. A
-- checkpointing job stores the offset and its state together, as one row.
CREATE TABLE job_progress (
  job         text   PRIMARY KEY,
  next_offset bigint NOT NULL,
  state       jsonb
);