harden versioned event substrate

internal/store/store.go

@@ -166,6 +166,9 @@ func (s *Store) Append(e domain.Event) error {
 	if e.Seq == 0 {
 		e.Seq = s.seq + 1
 	}
+	if e.SchemaVersion == 0 {
+		e.SchemaVersion = domain.CurrentEventSchema
+	}
 	if e.Type == "TaskCreated" {
 		var p map[string]any
 		if err := json.Unmarshal(e.Payload, &p); err != nil {
@@ -175,10 +178,22 @@ func (s *Store) Append(e domain.Event) error {
 			return nil
 		}
 	}
-	if t, ok := s.tasks[e.TaskID]; ok && e.Version != t.Version+1 {
+	t, taskExists := s.tasks[e.TaskID]
+	if taskExists && e.Version != t.Version+1 {
 		return domain.ErrConflict
 	}
-	if _, ok := s.tasks[e.TaskID]; !ok && e.Type != "TaskCreated" {
+	if e.Type == "TaskLeased" {
+		var p struct {
+			ExpectedVersion *int `json:"expected_version"`
+		}
+		if err := json.Unmarshal(e.Payload, &p); err != nil {
+			return err
+		}
+		if p.ExpectedVersion != nil && (t.Version != *p.ExpectedVersion) {
+			return domain.ErrConflict
+		}
+	}
+	if !taskExists && e.Type != "TaskCreated" {
 		return domain.ErrNotFound
 	}
 	if e.Type != "TaskCreated" && (e.Type == "TaskCompleted" || e.Type == "TaskBlocked" || e.Type == "TaskReleased") {
@@ -293,7 +308,7 @@ func (s *Store) Lease(id, harness string, ttl time.Duration) (domain.Event, erro
 	if t.State != domain.StateQueued {
 		return domain.Event{}, domain.ErrConflict
 	}
-	p, _ := json.Marshal(map[string]any{"harness_id": harness, "until_ns": time.Now().Add(ttl).UnixNano()})
+	p, _ := json.Marshal(map[string]any{"harness_id": harness, "until_ns": time.Now().Add(ttl).UnixNano(), "expected_version": t.Version})
 	e := domain.Event{ID: id, Type: "TaskLeased", TaskID: id, Version: t.Version + 1, Payload: p}
 	return e, s.Append(e)
 }