detached 38857c6

feat(kvstore): persist issue history and the resolved state, migrate to schema 8

commit
38857c63e2a85cd3b30e20c6f0c88202c731796c
author
brice · · 2026-10-22 18:13 UTC
committer
brice · · 2026-10-22 18:41 UTC
parent
7288afe
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
7 files · +484 -6
Unified
UnifiedSplit
M docs/storage.md +18 -0
@@ -90,3 +90,21 @@ service, use the same environment variables with its existing bucket, endpoint,
9090 region and keys, and `docker compose up ermite` without the profile. These are
9191 development settings; never use the placeholder admin token or MinIO keys in
9292 production.
93+
94+## Issue history (schema 8)
95+
96+Issue events are protobuf `IssueEvent` records under
97+`isev/<repo>/<issue:010d>/<seq:010d>`. Sequences are dense, per-issue and
98+1-based. Events naming a merge request are copied verbatim to
99+`mrev/<repo>/<mr:010d>/<issue:010d>/<seq:010d>` in the same atomic batch as
100+the issue update or link diff. Issue history is read in sequence order; MR
101+history is ordered by timestamp, then issue and sequence (at most 500 rows).
102+Each issue retains at most 500 events; further events are silently dropped
103+without preventing state or link updates. Removing a link preserves history;
104+deleting its repository removes both prefixes.
105+
106+Schema 8 adds the resolved issue state and migrates closed issues with a
107+nonzero `closed_by_merge_request` to resolved, without changing timestamps,
108+open indexes, or creating retroactive events. The migration and schema bump
109+are atomic and idempotent. Like token upgrades, this uses a single batch
110+(16 MiB limit, typically over 10k issue records). Older builds refuse schema 8.
@@ -90,3 +90,21 @@ service, use the same environment variables with its existing bucket, endpoint,
90 region and keys, and `docker compose up ermite` without the profile. These are90 region and keys, and `docker compose up ermite` without the profile. These are
91 development settings; never use the placeholder admin token or MinIO keys in91 development settings; never use the placeholder admin token or MinIO keys in
92 production.92 production.
93+
94+## Issue history (schema 8)
95+
96+Issue events are protobuf `IssueEvent` records under
97+`isev/<repo>/<issue:010d>/<seq:010d>`. Sequences are dense, per-issue and
98+1-based. Events naming a merge request are copied verbatim to
99+`mrev/<repo>/<mr:010d>/<issue:010d>/<seq:010d>` in the same atomic batch as
100+the issue update or link diff. Issue history is read in sequence order; MR
101+history is ordered by timestamp, then issue and sequence (at most 500 rows).
102+Each issue retains at most 500 events; further events are silently dropped
103+without preventing state or link updates. Removing a link preserves history;
104+deleting its repository removes both prefixes.
105+
106+Schema 8 adds the resolved issue state and migrates closed issues with a
107+nonzero `closed_by_merge_request` to resolved, without changing timestamps,
108+open indexes, or creating retroactive events. The migration and schema bump
109+are atomic and idempotent. Like token upgrades, this uses a single batch
110+(16 MiB limit, typically over 10k issue records). Older builds refuse schema 8.
A internal/adapters/kvstore/issue_history_internal_test.go +39 -0
@@ -0,0 +1,39 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+package kvstore
3+
4+import (
5+ "context"
6+ "testing"
7+
8+ kv "bnema.dev/ermite-kv"
9+ "bnema.dev/ermite/internal/domain"
10+ "github.com/stretchr/testify/require"
11+)
12+
13+func TestPreparedEventSequenceConflict(t *testing.T) {
14+ ctx := context.Background()
15+ db, err := kv.Open(ctx, t.TempDir(), kv.Options{})
16+ require.NoError(t, err)
17+ defer db.Close(ctx)
18+ repo, err := domain.ParseRepoName("proj")
19+ require.NoError(t, err)
20+ events := []domain.IssueEvent{{Kind: domain.IssueEventClosed, MergeRequest: 9}}
21+ var a, b kv.Batch
22+ require.NoError(t, prepareIssueEvents(ctx, db, &a, repo, 1, events))
23+ require.NoError(t, prepareIssueEvents(ctx, db, &b, repo, 1, events))
24+ b.Put([]byte("unrelated"), []byte{1})
25+ _, err = db.Apply(ctx, &a)
26+ require.NoError(t, err)
27+ _, err = db.Apply(ctx, &b)
28+ require.ErrorIs(t, err, kv.ErrConflict)
29+ _, exists := db.Get(ctx, []byte("unrelated"))
30+ require.False(t, exists)
31+ var retry kv.Batch
32+ require.NoError(t, prepareIssueEvents(ctx, db, &retry, repo, 1, events))
33+ _, err = db.Apply(ctx, &retry)
34+ require.NoError(t, err)
35+ history, err := (&Issues{db: db}).ListEvents(ctx, repo, 1)
36+ require.NoError(t, err)
37+ require.Len(t, history, 2)
38+ require.Equal(t, int64(2), history[1].ID)
39+}
@@ -0,0 +1,39 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+package kvstore
3+
4+import (
5+ "context"
6+ "testing"
7+
8+ kv "bnema.dev/ermite-kv"
9+ "bnema.dev/ermite/internal/domain"
10+ "github.com/stretchr/testify/require"
11+)
12+
13+func TestPreparedEventSequenceConflict(t *testing.T) {
14+ ctx := context.Background()
15+ db, err := kv.Open(ctx, t.TempDir(), kv.Options{})
16+ require.NoError(t, err)
17+ defer db.Close(ctx)
18+ repo, err := domain.ParseRepoName("proj")
19+ require.NoError(t, err)
20+ events := []domain.IssueEvent{{Kind: domain.IssueEventClosed, MergeRequest: 9}}
21+ var a, b kv.Batch
22+ require.NoError(t, prepareIssueEvents(ctx, db, &a, repo, 1, events))
23+ require.NoError(t, prepareIssueEvents(ctx, db, &b, repo, 1, events))
24+ b.Put([]byte("unrelated"), []byte{1})
25+ _, err = db.Apply(ctx, &a)
26+ require.NoError(t, err)
27+ _, err = db.Apply(ctx, &b)
28+ require.ErrorIs(t, err, kv.ErrConflict)
29+ _, exists := db.Get(ctx, []byte("unrelated"))
30+ require.False(t, exists)
31+ var retry kv.Batch
32+ require.NoError(t, prepareIssueEvents(ctx, db, &retry, repo, 1, events))
33+ _, err = db.Apply(ctx, &retry)
34+ require.NoError(t, err)
35+ history, err := (&Issues{db: db}).ListEvents(ctx, repo, 1)
36+ require.NoError(t, err)
37+ require.Len(t, history, 2)
38+ require.Equal(t, int64(2), history[1].ID)
39+}
A internal/adapters/kvstore/issue_history_test.go +226 -0
@@ -0,0 +1,226 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+package kvstore_test
3+
4+import (
5+ "context"
6+ "fmt"
7+ "testing"
8+ "time"
9+
10+ storagepb "bnema.dev/ermite/gen/ermite/storage"
11+ "bnema.dev/ermite/internal/adapters/kvstore"
12+ "bnema.dev/ermite/internal/domain"
13+ "bnema.dev/ermite/internal/ports"
14+ "github.com/stretchr/testify/assert"
15+ "github.com/stretchr/testify/require"
16+)
17+
18+func TestIssueHistoryAtomicAndCapped(t *testing.T) {
19+ ctx := context.Background()
20+ s, _, repo := repoWithMRs(t)
21+ is := s.Issues()
22+ issue, err := is.Create(ctx, newIssue(t, repo, "history"))
23+ require.NoError(t, err)
24+ stale := issue
25+ issue.State = domain.IssueStateResolved
26+ ev := domain.IssueEvent{Kind: domain.IssueEventResolved, Actor: "merger", MergeRequest: 9, CreatedAt: created}
27+ require.NoError(t, is.Update(ctx, issue, ev, domain.IssueEvent{Kind: domain.IssueEventReopened, CreatedAt: created.Add(time.Second)}))
28+ assert.ErrorIs(t, is.Update(ctx, stale, ev), domain.ErrConflict)
29+ events, err := is.ListEvents(ctx, repo, issue.Number)
30+ require.NoError(t, err)
31+ require.Len(t, events, 2)
32+ assert.Equal(t, int64(1), events[0].ID)
33+ assert.Equal(t, int64(2), events[1].ID)
34+ assert.Equal(t, issue.Number, events[0].Issue)
35+ assert.Equal(t, "merger", events[0].Actor)
36+ open, err := is.ListOpenAll(ctx)
37+ require.NoError(t, err)
38+ assert.Empty(t, open)
39+ page, err := is.List(ctx, repo, ports.ListIssuesParams{State: domain.IssueStateResolved})
40+ require.NoError(t, err)
41+ require.Len(t, page.Issues, 1)
42+ issue = page.Issues[0]
43+ batch := make([]domain.IssueEvent, kvstore.MaxIssueEvents)
44+ for j := range batch {
45+ batch[j] = ev
46+ }
47+ require.NoError(t, is.Update(ctx, issue, batch...))
48+ issue, err = is.Get(ctx, repo, issue.Number)
49+ require.NoError(t, err)
50+ issue.State = domain.IssueStateOpen
51+ require.NoError(t, is.Update(ctx, issue, ev))
52+ events, err = is.ListEvents(ctx, repo, issue.Number)
53+ require.NoError(t, err)
54+ require.Len(t, events, kvstore.MaxIssueEvents)
55+ for j, e := range events {
56+ assert.Equal(t, int64(j+1), e.ID)
57+ }
58+ issue, err = is.Get(ctx, repo, issue.Number)
59+ require.NoError(t, err)
60+ assert.Equal(t, domain.IssueStateOpen, issue.State)
61+ require.NoError(t, s.Delete(ctx, repo))
62+ for _, prefix := range []string{"is/", "isc/", "isopen/", "isev/", "mrev/"} {
63+ assert.Zero(t, countKeys(t, s, prefix+repo.String()+"/"))
64+ }
65+}
66+
67+func TestReferenceHistoryDiff(t *testing.T) {
68+ ctx := context.Background()
69+ s, _, repo := repoWithMRs(t)
70+ links := s.IssueLinks()
71+ ref := ports.ReferenceEvent{Actor: "writer", At: created}
72+ replace := func(issues, closes []int64) {
73+ t.Helper()
74+ _, _, err := links.ReplaceLinks(ctx, repo, 9, issues, closes, ref)
75+ require.NoError(t, err)
76+ }
77+ replace([]int64{3}, nil)
78+ replace([]int64{3}, nil)
79+ ref.At = created.Add(time.Second)
80+ replace([]int64{3}, []int64{3})
81+ replace([]int64{3}, []int64{3})
82+ replace([]int64{3}, nil) // downgrade has no event
83+ replace(nil, nil) // removal retains history
84+ events, err := s.Issues().ListEvents(ctx, repo, 3)
85+ require.NoError(t, err)
86+ require.Len(t, events, 2)
87+ assert.False(t, events[0].Closes)
88+ assert.True(t, events[1].Closes)
89+ assert.Equal(t, "writer", events[0].Actor)
90+ assert.Equal(t, domain.IssueEventReferenced, events[0].Kind)
91+ mrEvents, err := links.EventsOfMergeRequest(ctx, repo, 9)
92+ require.NoError(t, err)
93+ assert.Equal(t, events, mrEvents)
94+ replace([]int64{3}, nil) // reappearance is another event
95+ events, err = s.Issues().ListEvents(ctx, repo, 3)
96+ require.NoError(t, err)
97+ require.Len(t, events, 3)
98+ assert.Equal(t, int64(3), events[2].ID)
99+ require.NoError(t, s.Delete(ctx, repo))
100+ assert.Zero(t, countKeys(t, s, "isev/"+repo.String()+"/"))
101+ assert.Zero(t, countKeys(t, s, "mrev/"+repo.String()+"/"))
102+}
103+
104+func TestMRHistoryOrderAndCap(t *testing.T) {
105+ ctx := context.Background()
106+ s, _, repo := repoWithMRs(t)
107+ // Insert in reverse timestamp order; issue and seq break timestamp ties.
108+ for n := int64(1); n <= 3; n++ {
109+ issue, err := s.Issues().Create(ctx, newIssue(t, repo, "many"))
110+ require.NoError(t, err)
111+ events := make([]domain.IssueEvent, 201)
112+ for j := range events {
113+ events[j] = domain.IssueEvent{Kind: domain.IssueEventResolved, MergeRequest: 9, CreatedAt: created.Add(time.Duration(200-j) * time.Second)}
114+ }
115+ require.NoError(t, s.Issues().Update(ctx, issue, events...))
116+ }
117+ events, err := s.IssueLinks().EventsOfMergeRequest(ctx, repo, 9)
118+ require.NoError(t, err)
119+ require.Len(t, events, 500)
120+ for j := 1; j < len(events); j++ {
121+ a, b := events[j-1], events[j]
122+ assert.False(t, a.CreatedAt.After(b.CreatedAt))
123+ if a.CreatedAt.Equal(b.CreatedAt) {
124+ assert.Less(t, a.Issue, b.Issue)
125+ }
126+ }
127+}
128+
129+func TestResolvedMigrationAndRestart(t *testing.T) {
130+ ctx := context.Background()
131+ for _, version := range []uint32{5, 7} {
132+ t.Run(fmt.Sprint(version), func(t *testing.T) {
133+ dir := t.TempDir()
134+ putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: version}))
135+ for n := int64(1); n <= 3; n++ {
136+ state := storagepb.IssueState_ISSUE_STATE_CLOSED
137+ mr := int64(0)
138+ if n == 1 {
139+ mr = 9
140+ }
141+ if n == 3 {
142+ state = storagepb.IssueState_ISSUE_STATE_OPEN
143+ }
144+ putRaw(t, dir, fmt.Sprintf("is/proj/%010d", n), mustProto(t, &storagepb.Issue{Number: n, Title: "old", State: state, ClosedByMergeRequest: mr}))
145+ }
146+ putRaw(t, dir, "isopen/proj/0000000003", []byte{1})
147+ repo := name(t, "proj")
148+ var migratedVersion uint64
149+ for pass := 0; pass < 2; pass++ {
150+ s, err := kvstore.Open(ctx, dir)
151+ require.NoError(t, err)
152+ for n, state := range []domain.IssueState{domain.IssueStateResolved, domain.IssueStateClosed, domain.IssueStateOpen} {
153+ issue, err := s.Issues().Get(ctx, repo, int64(n+1))
154+ require.NoError(t, err)
155+ assert.Equal(t, state, issue.State)
156+ assert.True(t, issue.UpdatedAt.IsZero())
157+ if n == 0 {
158+ assert.Equal(t, int64(9), issue.ClosedByMergeRequest)
159+ if pass == 0 {
160+ migratedVersion = issue.Version
161+ } else {
162+ assert.Equal(t, migratedVersion, issue.Version)
163+ }
164+ }
165+ }
166+ assert.Equal(t, 1, countKeys(t, s, "isopen/proj/"))
167+ assert.Zero(t, countKeys(t, s, "isev/"))
168+ assert.Zero(t, countKeys(t, s, "mrev/"))
169+ require.NoError(t, s.Close(ctx))
170+ assert.Equal(t, uint32(8), readSchema(t, dir))
171+ }
172+ })
173+ }
174+}
175+
176+func TestIssueEventSequenceSurvivesRestart(t *testing.T) {
177+ ctx := context.Background()
178+ dir := t.TempDir()
179+ repo := name(t, "proj")
180+ for pass := 0; pass < 2; pass++ {
181+ s, err := kvstore.Open(ctx, dir)
182+ require.NoError(t, err)
183+ if pass == 0 {
184+ require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate}))
185+ _, err := s.Issues().Create(ctx, newIssue(t, repo, "restart"))
186+ require.NoError(t, err)
187+ }
188+ issue, err := s.Issues().Get(ctx, repo, 1)
189+ require.NoError(t, err)
190+ require.NoError(t, s.Issues().Update(ctx, issue, domain.IssueEvent{Kind: domain.IssueEventClosed, MergeRequest: 9, CreatedAt: created}))
191+ events, err := s.IssueLinks().EventsOfMergeRequest(ctx, repo, 9)
192+ require.NoError(t, err)
193+ require.Len(t, events, pass+1)
194+ assert.Equal(t, int64(pass+1), events[pass].ID)
195+ require.NoError(t, s.Close(ctx))
196+ }
197+}
198+
199+func TestLinkEventsDropAtCapWithoutChangingIssueVersion(t *testing.T) {
200+ ctx := context.Background()
201+ s, _, repo := repoWithMRs(t)
202+ issue, err := s.Issues().Create(ctx, newIssue(t, repo, "full"))
203+ require.NoError(t, err)
204+ events := make([]domain.IssueEvent, kvstore.MaxIssueEvents)
205+ for j := range events {
206+ events[j].Kind = domain.IssueEventClosed
207+ }
208+ require.NoError(t, s.Issues().Update(ctx, issue, events...))
209+ before, err := s.Issues().Get(ctx, repo, issue.Number)
210+ require.NoError(t, err)
211+ added, _, err := s.IssueLinks().ReplaceLinks(ctx, repo, 9, []int64{issue.Number}, []int64{issue.Number}, ports.ReferenceEvent{Actor: "writer", At: created})
212+ require.NoError(t, err)
213+ require.Len(t, added, 1)
214+ after, err := s.Issues().Get(ctx, repo, issue.Number)
215+ require.NoError(t, err)
216+ assert.Equal(t, before.Version, after.Version)
217+ history, err := s.Issues().ListEvents(ctx, repo, issue.Number)
218+ require.NoError(t, err)
219+ require.Len(t, history, kvstore.MaxIssueEvents)
220+ mrHistory, err := s.IssueLinks().EventsOfMergeRequest(ctx, repo, 9)
221+ require.NoError(t, err)
222+ assert.Empty(t, mrHistory)
223+ links, err := s.IssueLinks().LinksOfMergeRequest(ctx, repo, 9)
224+ require.NoError(t, err)
225+ assert.Equal(t, added, links)
226+}
@@ -0,0 +1,226 @@
1+// SPDX-License-Identifier: AGPL-3.0-or-later
2+package kvstore_test
3+
4+import (
5+ "context"
6+ "fmt"
7+ "testing"
8+ "time"
9+
10+ storagepb "bnema.dev/ermite/gen/ermite/storage"
11+ "bnema.dev/ermite/internal/adapters/kvstore"
12+ "bnema.dev/ermite/internal/domain"
13+ "bnema.dev/ermite/internal/ports"
14+ "github.com/stretchr/testify/assert"
15+ "github.com/stretchr/testify/require"
16+)
17+
18+func TestIssueHistoryAtomicAndCapped(t *testing.T) {
19+ ctx := context.Background()
20+ s, _, repo := repoWithMRs(t)
21+ is := s.Issues()
22+ issue, err := is.Create(ctx, newIssue(t, repo, "history"))
23+ require.NoError(t, err)
24+ stale := issue
25+ issue.State = domain.IssueStateResolved
26+ ev := domain.IssueEvent{Kind: domain.IssueEventResolved, Actor: "merger", MergeRequest: 9, CreatedAt: created}
27+ require.NoError(t, is.Update(ctx, issue, ev, domain.IssueEvent{Kind: domain.IssueEventReopened, CreatedAt: created.Add(time.Second)}))
28+ assert.ErrorIs(t, is.Update(ctx, stale, ev), domain.ErrConflict)
29+ events, err := is.ListEvents(ctx, repo, issue.Number)
30+ require.NoError(t, err)
31+ require.Len(t, events, 2)
32+ assert.Equal(t, int64(1), events[0].ID)
33+ assert.Equal(t, int64(2), events[1].ID)
34+ assert.Equal(t, issue.Number, events[0].Issue)
35+ assert.Equal(t, "merger", events[0].Actor)
36+ open, err := is.ListOpenAll(ctx)
37+ require.NoError(t, err)
38+ assert.Empty(t, open)
39+ page, err := is.List(ctx, repo, ports.ListIssuesParams{State: domain.IssueStateResolved})
40+ require.NoError(t, err)
41+ require.Len(t, page.Issues, 1)
42+ issue = page.Issues[0]
43+ batch := make([]domain.IssueEvent, kvstore.MaxIssueEvents)
44+ for j := range batch {
45+ batch[j] = ev
46+ }
47+ require.NoError(t, is.Update(ctx, issue, batch...))
48+ issue, err = is.Get(ctx, repo, issue.Number)
49+ require.NoError(t, err)
50+ issue.State = domain.IssueStateOpen
51+ require.NoError(t, is.Update(ctx, issue, ev))
52+ events, err = is.ListEvents(ctx, repo, issue.Number)
53+ require.NoError(t, err)
54+ require.Len(t, events, kvstore.MaxIssueEvents)
55+ for j, e := range events {
56+ assert.Equal(t, int64(j+1), e.ID)
57+ }
58+ issue, err = is.Get(ctx, repo, issue.Number)
59+ require.NoError(t, err)
60+ assert.Equal(t, domain.IssueStateOpen, issue.State)
61+ require.NoError(t, s.Delete(ctx, repo))
62+ for _, prefix := range []string{"is/", "isc/", "isopen/", "isev/", "mrev/"} {
63+ assert.Zero(t, countKeys(t, s, prefix+repo.String()+"/"))
64+ }
65+}
66+
67+func TestReferenceHistoryDiff(t *testing.T) {
68+ ctx := context.Background()
69+ s, _, repo := repoWithMRs(t)
70+ links := s.IssueLinks()
71+ ref := ports.ReferenceEvent{Actor: "writer", At: created}
72+ replace := func(issues, closes []int64) {
73+ t.Helper()
74+ _, _, err := links.ReplaceLinks(ctx, repo, 9, issues, closes, ref)
75+ require.NoError(t, err)
76+ }
77+ replace([]int64{3}, nil)
78+ replace([]int64{3}, nil)
79+ ref.At = created.Add(time.Second)
80+ replace([]int64{3}, []int64{3})
81+ replace([]int64{3}, []int64{3})
82+ replace([]int64{3}, nil) // downgrade has no event
83+ replace(nil, nil) // removal retains history
84+ events, err := s.Issues().ListEvents(ctx, repo, 3)
85+ require.NoError(t, err)
86+ require.Len(t, events, 2)
87+ assert.False(t, events[0].Closes)
88+ assert.True(t, events[1].Closes)
89+ assert.Equal(t, "writer", events[0].Actor)
90+ assert.Equal(t, domain.IssueEventReferenced, events[0].Kind)
91+ mrEvents, err := links.EventsOfMergeRequest(ctx, repo, 9)
92+ require.NoError(t, err)
93+ assert.Equal(t, events, mrEvents)
94+ replace([]int64{3}, nil) // reappearance is another event
95+ events, err = s.Issues().ListEvents(ctx, repo, 3)
96+ require.NoError(t, err)
97+ require.Len(t, events, 3)
98+ assert.Equal(t, int64(3), events[2].ID)
99+ require.NoError(t, s.Delete(ctx, repo))
100+ assert.Zero(t, countKeys(t, s, "isev/"+repo.String()+"/"))
101+ assert.Zero(t, countKeys(t, s, "mrev/"+repo.String()+"/"))
102+}
103+
104+func TestMRHistoryOrderAndCap(t *testing.T) {
105+ ctx := context.Background()
106+ s, _, repo := repoWithMRs(t)
107+ // Insert in reverse timestamp order; issue and seq break timestamp ties.
108+ for n := int64(1); n <= 3; n++ {
109+ issue, err := s.Issues().Create(ctx, newIssue(t, repo, "many"))
110+ require.NoError(t, err)
111+ events := make([]domain.IssueEvent, 201)
112+ for j := range events {
113+ events[j] = domain.IssueEvent{Kind: domain.IssueEventResolved, MergeRequest: 9, CreatedAt: created.Add(time.Duration(200-j) * time.Second)}
114+ }
115+ require.NoError(t, s.Issues().Update(ctx, issue, events...))
116+ }
117+ events, err := s.IssueLinks().EventsOfMergeRequest(ctx, repo, 9)
118+ require.NoError(t, err)
119+ require.Len(t, events, 500)
120+ for j := 1; j < len(events); j++ {
121+ a, b := events[j-1], events[j]
122+ assert.False(t, a.CreatedAt.After(b.CreatedAt))
123+ if a.CreatedAt.Equal(b.CreatedAt) {
124+ assert.Less(t, a.Issue, b.Issue)
125+ }
126+ }
127+}
128+
129+func TestResolvedMigrationAndRestart(t *testing.T) {
130+ ctx := context.Background()
131+ for _, version := range []uint32{5, 7} {
132+ t.Run(fmt.Sprint(version), func(t *testing.T) {
133+ dir := t.TempDir()
134+ putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: version}))
135+ for n := int64(1); n <= 3; n++ {
136+ state := storagepb.IssueState_ISSUE_STATE_CLOSED
137+ mr := int64(0)
138+ if n == 1 {
139+ mr = 9
140+ }
141+ if n == 3 {
142+ state = storagepb.IssueState_ISSUE_STATE_OPEN
143+ }
144+ putRaw(t, dir, fmt.Sprintf("is/proj/%010d", n), mustProto(t, &storagepb.Issue{Number: n, Title: "old", State: state, ClosedByMergeRequest: mr}))
145+ }
146+ putRaw(t, dir, "isopen/proj/0000000003", []byte{1})
147+ repo := name(t, "proj")
148+ var migratedVersion uint64
149+ for pass := 0; pass < 2; pass++ {
150+ s, err := kvstore.Open(ctx, dir)
151+ require.NoError(t, err)
152+ for n, state := range []domain.IssueState{domain.IssueStateResolved, domain.IssueStateClosed, domain.IssueStateOpen} {
153+ issue, err := s.Issues().Get(ctx, repo, int64(n+1))
154+ require.NoError(t, err)
155+ assert.Equal(t, state, issue.State)
156+ assert.True(t, issue.UpdatedAt.IsZero())
157+ if n == 0 {
158+ assert.Equal(t, int64(9), issue.ClosedByMergeRequest)
159+ if pass == 0 {
160+ migratedVersion = issue.Version
161+ } else {
162+ assert.Equal(t, migratedVersion, issue.Version)
163+ }
164+ }
165+ }
166+ assert.Equal(t, 1, countKeys(t, s, "isopen/proj/"))
167+ assert.Zero(t, countKeys(t, s, "isev/"))
168+ assert.Zero(t, countKeys(t, s, "mrev/"))
169+ require.NoError(t, s.Close(ctx))
170+ assert.Equal(t, uint32(8), readSchema(t, dir))
171+ }
172+ })
173+ }
174+}
175+
176+func TestIssueEventSequenceSurvivesRestart(t *testing.T) {
177+ ctx := context.Background()
178+ dir := t.TempDir()
179+ repo := name(t, "proj")
180+ for pass := 0; pass < 2; pass++ {
181+ s, err := kvstore.Open(ctx, dir)
182+ require.NoError(t, err)
183+ if pass == 0 {
184+ require.NoError(t, s.Create(ctx, domain.Repo{Name: repo, Visibility: domain.VisibilityPrivate}))
185+ _, err := s.Issues().Create(ctx, newIssue(t, repo, "restart"))
186+ require.NoError(t, err)
187+ }
188+ issue, err := s.Issues().Get(ctx, repo, 1)
189+ require.NoError(t, err)
190+ require.NoError(t, s.Issues().Update(ctx, issue, domain.IssueEvent{Kind: domain.IssueEventClosed, MergeRequest: 9, CreatedAt: created}))
191+ events, err := s.IssueLinks().EventsOfMergeRequest(ctx, repo, 9)
192+ require.NoError(t, err)
193+ require.Len(t, events, pass+1)
194+ assert.Equal(t, int64(pass+1), events[pass].ID)
195+ require.NoError(t, s.Close(ctx))
196+ }
197+}
198+
199+func TestLinkEventsDropAtCapWithoutChangingIssueVersion(t *testing.T) {
200+ ctx := context.Background()
201+ s, _, repo := repoWithMRs(t)
202+ issue, err := s.Issues().Create(ctx, newIssue(t, repo, "full"))
203+ require.NoError(t, err)
204+ events := make([]domain.IssueEvent, kvstore.MaxIssueEvents)
205+ for j := range events {
206+ events[j].Kind = domain.IssueEventClosed
207+ }
208+ require.NoError(t, s.Issues().Update(ctx, issue, events...))
209+ before, err := s.Issues().Get(ctx, repo, issue.Number)
210+ require.NoError(t, err)
211+ added, _, err := s.IssueLinks().ReplaceLinks(ctx, repo, 9, []int64{issue.Number}, []int64{issue.Number}, ports.ReferenceEvent{Actor: "writer", At: created})
212+ require.NoError(t, err)
213+ require.Len(t, added, 1)
214+ after, err := s.Issues().Get(ctx, repo, issue.Number)
215+ require.NoError(t, err)
216+ assert.Equal(t, before.Version, after.Version)
217+ history, err := s.Issues().ListEvents(ctx, repo, issue.Number)
218+ require.NoError(t, err)
219+ require.Len(t, history, kvstore.MaxIssueEvents)
220+ mrHistory, err := s.IssueLinks().EventsOfMergeRequest(ctx, repo, 9)
221+ require.NoError(t, err)
222+ assert.Empty(t, mrHistory)
223+ links, err := s.IssueLinks().LinksOfMergeRequest(ctx, repo, 9)
224+ require.NoError(t, err)
225+ assert.Equal(t, added, links)
226+}
M internal/adapters/kvstore/issue_links.go +49 -2
@@ -91,6 +91,11 @@ func (l *IssueLinks) ReplaceLinks(ctx context.Context, repo domain.RepoName, mr
9191 added = append(added, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: closing})
9292 }
9393 }
94+ for _, link := range added {
95+ if err := prepareIssueEvents(ctx, l.db, &b, repo, link.Issue, []domain.IssueEvent{{Kind: domain.IssueEventReferenced, MergeRequest: mr, Closes: link.Closes, Actor: ref.Actor, CreatedAt: ref.At}}); err != nil {
96+ return nil, nil, err
97+ }
98+ }
9499 b.Put(key, mustMarshal(&storagepb.MergeRequestReferences{Issues: issues, Closes: closes}))
95100 if _, err := l.db.Apply(ctx, &b); err != nil {
96101 if errors.Is(err, kv.ErrConflict) {
@@ -150,7 +155,49 @@ func toSet(numbers []int64) map[int64]struct{} {
150155 return set
151156 }
152157
153-// EventsOfMergeRequest returns MR issue history. Persistence is implemented in Phase A.
158+// EventsOfMergeRequest returns history, including references no longer linked,
159+// ordered by time, then issue and sequence, capped at MaxIssueEvents.
154160 func (l *IssueLinks) EventsOfMergeRequest(ctx context.Context, repo domain.RepoName, mr int64) ([]domain.IssueEvent, error) {
155- return nil, nil
161+ var out []domain.IssueEvent
162+ prefix := mrEventsPrefix(repo, mr)
163+ for key, item := range l.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
164+ if err := ctx.Err(); err != nil {
165+ return nil, err
166+ }
167+ num, _, ok := strings.Cut(strings.TrimPrefix(string(key), prefix), "/")
168+ if !ok {
169+ return nil, fmt.Errorf("invalid MR event key %q", key)
170+ }
171+ issue, err := strconv.ParseInt(num, 10, 64)
172+ if err != nil {
173+ return nil, fmt.Errorf("decode MR event key: %w", err)
174+ }
175+ ev, err := decodeIssueEvent(item.Value, issue)
176+ if err != nil {
177+ return nil, err
178+ }
179+ out = append(out, ev)
180+ }
181+ slices.SortFunc(out, func(a, b domain.IssueEvent) int {
182+ if c := a.CreatedAt.Compare(b.CreatedAt); c != 0 {
183+ return c
184+ }
185+ if a.Issue < b.Issue {
186+ return -1
187+ }
188+ if a.Issue > b.Issue {
189+ return 1
190+ }
191+ if a.ID < b.ID {
192+ return -1
193+ }
194+ if a.ID > b.ID {
195+ return 1
196+ }
197+ return 0
198+ })
199+ if len(out) > MaxIssueEvents {
200+ out = out[:MaxIssueEvents]
201+ }
202+ return out, ctx.Err()
156203 }
@@ -91,6 +91,11 @@ func (l *IssueLinks) ReplaceLinks(ctx context.Context, repo domain.RepoName, mr
91 added = append(added, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: closing})91 added = append(added, domain.IssueLink{Issue: n, MergeRequest: mr, Closes: closing})
92 }92 }
93 }93 }
94+ for _, link := range added {
95+ if err := prepareIssueEvents(ctx, l.db, &b, repo, link.Issue, []domain.IssueEvent{{Kind: domain.IssueEventReferenced, MergeRequest: mr, Closes: link.Closes, Actor: ref.Actor, CreatedAt: ref.At}}); err != nil {
96+ return nil, nil, err
97+ }
98+ }
94 b.Put(key, mustMarshal(&storagepb.MergeRequestReferences{Issues: issues, Closes: closes}))99 b.Put(key, mustMarshal(&storagepb.MergeRequestReferences{Issues: issues, Closes: closes}))
95 if _, err := l.db.Apply(ctx, &b); err != nil {100 if _, err := l.db.Apply(ctx, &b); err != nil {
96 if errors.Is(err, kv.ErrConflict) {101 if errors.Is(err, kv.ErrConflict) {
@@ -150,7 +155,49 @@ func toSet(numbers []int64) map[int64]struct{} {
150 return set155 return set
151 }156 }
152 157
153-// EventsOfMergeRequest returns MR issue history. Persistence is implemented in Phase A.158+// EventsOfMergeRequest returns history, including references no longer linked,
159+// ordered by time, then issue and sequence, capped at MaxIssueEvents.
154 func (l *IssueLinks) EventsOfMergeRequest(ctx context.Context, repo domain.RepoName, mr int64) ([]domain.IssueEvent, error) {160 func (l *IssueLinks) EventsOfMergeRequest(ctx context.Context, repo domain.RepoName, mr int64) ([]domain.IssueEvent, error) {
155- return nil, nil161+ var out []domain.IssueEvent
162+ prefix := mrEventsPrefix(repo, mr)
163+ for key, item := range l.db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
164+ if err := ctx.Err(); err != nil {
165+ return nil, err
166+ }
167+ num, _, ok := strings.Cut(strings.TrimPrefix(string(key), prefix), "/")
168+ if !ok {
169+ return nil, fmt.Errorf("invalid MR event key %q", key)
170+ }
171+ issue, err := strconv.ParseInt(num, 10, 64)
172+ if err != nil {
173+ return nil, fmt.Errorf("decode MR event key: %w", err)
174+ }
175+ ev, err := decodeIssueEvent(item.Value, issue)
176+ if err != nil {
177+ return nil, err
178+ }
179+ out = append(out, ev)
180+ }
181+ slices.SortFunc(out, func(a, b domain.IssueEvent) int {
182+ if c := a.CreatedAt.Compare(b.CreatedAt); c != 0 {
183+ return c
184+ }
185+ if a.Issue < b.Issue {
186+ return -1
187+ }
188+ if a.Issue > b.Issue {
189+ return 1
190+ }
191+ if a.ID < b.ID {
192+ return -1
193+ }
194+ if a.ID > b.ID {
195+ return 1
196+ }
197+ return 0
198+ })
199+ if len(out) > MaxIssueEvents {
200+ out = out[:MaxIssueEvents]
201+ }
202+ return out, ctx.Err()
156 }203 }
M internal/adapters/kvstore/issues.go +117 -2
@@ -23,6 +23,8 @@ import (
2323 // from the sequence of the merge requests (mrseq/<repo>): an issue and a merge
2424 // request never share one.
2525 const (
26+ issueEventPrefix = "isev/"
27+ mrEventPrefix = "mrev/"
2628 issuePrefix = "is/" // is/<repo>/<number:010d>
2729 issueCommentPrefix = "isc/" // isc/<repo>/<number:010d>/<id:010d>
2830 issueOpenPrefix = "isopen/" // isopen/<repo>/<number:010d>, written with the issue
@@ -72,6 +74,7 @@ func indexOpenIssue(b *kv.Batch, issue domain.Issue) {
7274 func issueKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
7375 var keys [][]byte
7476 for _, p := range []string{
77+ issueEventPrefix + repo.String() + "/", mrEventPrefix + repo.String() + "/",
7578 issueRepoPrefix(repo), issueCommentPrefix + repo.String() + "/", issueOpenRepoPrefix(repo), issueRefPrefix + repo.String() + "/",
7679 } {
7780 for k := range snap.Prefix(ctx, []byte(p)) {
@@ -240,6 +243,9 @@ func (i *Issues) Update(ctx context.Context, issue domain.Issue, events ...domai
240243 b.RequireVersion(key, issue.Version)
241244 b.Put(key, val)
242245 indexOpenIssue(&b, issue)
246+ if err := prepareIssueEvents(ctx, i.db, &b, issue.Repo, issue.Number, events); err != nil {
247+ return err
248+ }
243249 if _, err := i.db.Apply(ctx, &b); err != nil {
244250 if errors.Is(err, kv.ErrConflict) {
245251 return domain.ErrConflict
@@ -387,6 +393,8 @@ func encodeIssueState(s domain.IssueState) (storagepb.IssueState, error) {
387393 switch s {
388394 case domain.IssueStateOpen:
389395 return storagepb.IssueState_ISSUE_STATE_OPEN, nil
396+ case domain.IssueStateResolved:
397+ return storagepb.IssueState_ISSUE_STATE_RESOLVED, nil
390398 case domain.IssueStateClosed:
391399 return storagepb.IssueState_ISSUE_STATE_CLOSED, nil
392400 }
@@ -397,6 +405,8 @@ func decodeIssueState(s storagepb.IssueState) (domain.IssueState, error) {
397405 switch s {
398406 case storagepb.IssueState_ISSUE_STATE_OPEN:
399407 return domain.IssueStateOpen, nil
408+ case storagepb.IssueState_ISSUE_STATE_RESOLVED:
409+ return domain.IssueStateResolved, nil
400410 case storagepb.IssueState_ISSUE_STATE_CLOSED:
401411 return domain.IssueStateClosed, nil
402412 case storagepb.IssueState_ISSUE_STATE_PENDING:
@@ -434,7 +444,112 @@ func decodeIssuesMode(m storagepb.IssuesMode) domain.IssuesMode {
434444 // are silently dropped without preventing the associated write.
435445 const MaxIssueEvents = 500
436446
437-// ListEvents returns issue history. Persistence is implemented in Phase A.
447+// ListEvents returns the append-only issue history in sequence order.
438448 func (i *Issues) ListEvents(ctx context.Context, repo domain.RepoName, number int64) ([]domain.IssueEvent, error) {
439- return nil, nil
449+ var out []domain.IssueEvent
450+ for _, item := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueEventsPrefix(repo, number))) {
451+ if err := ctx.Err(); err != nil {
452+ return nil, err
453+ }
454+ ev, err := decodeIssueEvent(item.Value, number)
455+ if err != nil {
456+ return nil, err
457+ }
458+ out = append(out, ev)
459+ if len(out) == MaxIssueEvents {
460+ break
461+ }
462+ }
463+ return out, ctx.Err()
464+}
465+
466+func issueEventsPrefix(repo domain.RepoName, issue int64) string {
467+ return issueEventPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, issue) + "/"
468+}
469+
470+// prepareIssueEvents allocates dense per-issue sequences. RequireAbsent makes
471+// concurrent appenders conflict; callers retry the entire associated write.
472+// Scan once for a group so multiple events in one batch get distinct sequences.
473+func prepareIssueEvents(ctx context.Context, db *kv.DB, b *kv.Batch, repo domain.RepoName, issue int64, events []domain.IssueEvent) error {
474+ if len(events) == 0 {
475+ return nil
476+ }
477+ prefix := issueEventsPrefix(repo, issue)
478+ var seq int64
479+ count := 0
480+ for key := range db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
481+ if err := ctx.Err(); err != nil {
482+ return err
483+ }
484+ n, err := strconv.ParseInt(strings.TrimPrefix(string(key), prefix), 10, 64)
485+ if err != nil {
486+ return fmt.Errorf("decode issue event key: %w", err)
487+ }
488+ seq = max(seq, n)
489+ count++
490+ }
491+ for _, ev := range events {
492+ if count >= MaxIssueEvents {
493+ break
494+ }
495+ seq++
496+ ev.ID = seq
497+ val, err := encodeIssueEvent(ev)
498+ if err != nil {
499+ return err
500+ }
501+ key := []byte(prefix + fmt.Sprintf(numberFormat, seq))
502+ b.RequireAbsent(key)
503+ b.Put(key, val)
504+ if ev.MergeRequest != 0 {
505+ b.Put([]byte(mrEventsPrefix(repo, ev.MergeRequest)+fmt.Sprintf(numberFormat, issue)+"/"+fmt.Sprintf(numberFormat, seq)), val)
506+ }
507+ count++
508+ }
509+ return ctx.Err()
510+}
511+
512+func mrEventsPrefix(repo domain.RepoName, mr int64) string {
513+ return mrEventPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, mr) + "/"
514+}
515+
516+func encodeIssueEvent(ev domain.IssueEvent) ([]byte, error) {
517+ var kind storagepb.IssueEventKind
518+ switch ev.Kind {
519+ case domain.IssueEventReferenced:
520+ kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_REFERENCED
521+ case domain.IssueEventClosed:
522+ kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_CLOSED
523+ case domain.IssueEventReopened:
524+ kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_REOPENED
525+ case domain.IssueEventResolved:
526+ kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_RESOLVED
527+ default:
528+ return nil, fmt.Errorf("%w: issue event kind %q", domain.ErrInvalid, ev.Kind)
529+ }
530+ if !utf8.ValidString(ev.Actor) {
531+ return nil, errors.New("issue event actor is not valid UTF-8")
532+ }
533+ return mustMarshal(&storagepb.IssueEvent{Seq: ev.ID, Kind: kind, MergeRequest: ev.MergeRequest, Closes: ev.Closes, Actor: ev.Actor, CreatedAt: encodeTime(ev.CreatedAt)}), nil
534+}
535+
536+func decodeIssueEvent(val []byte, issue int64) (domain.IssueEvent, error) {
537+ var pb storagepb.IssueEvent
538+ if err := proto.Unmarshal(val, &pb); err != nil {
539+ return domain.IssueEvent{}, fmt.Errorf("decode issue event: %w", err)
540+ }
541+ var kind domain.IssueEventKind
542+ switch pb.GetKind() {
543+ case storagepb.IssueEventKind_ISSUE_EVENT_KIND_REFERENCED:
544+ kind = domain.IssueEventReferenced
545+ case storagepb.IssueEventKind_ISSUE_EVENT_KIND_CLOSED:
546+ kind = domain.IssueEventClosed
547+ case storagepb.IssueEventKind_ISSUE_EVENT_KIND_REOPENED:
548+ kind = domain.IssueEventReopened
549+ case storagepb.IssueEventKind_ISSUE_EVENT_KIND_RESOLVED:
550+ kind = domain.IssueEventResolved
551+ default:
552+ return domain.IssueEvent{}, fmt.Errorf("unknown issue event kind %d", pb.GetKind())
553+ }
554+ return domain.IssueEvent{ID: pb.GetSeq(), Issue: issue, Kind: kind, MergeRequest: pb.GetMergeRequest(), Closes: pb.GetCloses(), Actor: pb.GetActor(), CreatedAt: decodeTime(pb.GetCreatedAt())}, nil
440555 }
@@ -23,6 +23,8 @@ import (
23 // from the sequence of the merge requests (mrseq/<repo>): an issue and a merge23 // from the sequence of the merge requests (mrseq/<repo>): an issue and a merge
24 // request never share one.24 // request never share one.
25 const (25 const (
26+ issueEventPrefix = "isev/"
27+ mrEventPrefix = "mrev/"
26 issuePrefix = "is/" // is/<repo>/<number:010d>28 issuePrefix = "is/" // is/<repo>/<number:010d>
27 issueCommentPrefix = "isc/" // isc/<repo>/<number:010d>/<id:010d>29 issueCommentPrefix = "isc/" // isc/<repo>/<number:010d>/<id:010d>
28 issueOpenPrefix = "isopen/" // isopen/<repo>/<number:010d>, written with the issue30 issueOpenPrefix = "isopen/" // isopen/<repo>/<number:010d>, written with the issue
@@ -72,6 +74,7 @@ func indexOpenIssue(b *kv.Batch, issue domain.Issue) {
72 func issueKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {74 func issueKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
73 var keys [][]byte75 var keys [][]byte
74 for _, p := range []string{76 for _, p := range []string{
77+ issueEventPrefix + repo.String() + "/", mrEventPrefix + repo.String() + "/",
75 issueRepoPrefix(repo), issueCommentPrefix + repo.String() + "/", issueOpenRepoPrefix(repo), issueRefPrefix + repo.String() + "/",78 issueRepoPrefix(repo), issueCommentPrefix + repo.String() + "/", issueOpenRepoPrefix(repo), issueRefPrefix + repo.String() + "/",
76 } {79 } {
77 for k := range snap.Prefix(ctx, []byte(p)) {80 for k := range snap.Prefix(ctx, []byte(p)) {
@@ -240,6 +243,9 @@ func (i *Issues) Update(ctx context.Context, issue domain.Issue, events ...domai
240 b.RequireVersion(key, issue.Version)243 b.RequireVersion(key, issue.Version)
241 b.Put(key, val)244 b.Put(key, val)
242 indexOpenIssue(&b, issue)245 indexOpenIssue(&b, issue)
246+ if err := prepareIssueEvents(ctx, i.db, &b, issue.Repo, issue.Number, events); err != nil {
247+ return err
248+ }
243 if _, err := i.db.Apply(ctx, &b); err != nil {249 if _, err := i.db.Apply(ctx, &b); err != nil {
244 if errors.Is(err, kv.ErrConflict) {250 if errors.Is(err, kv.ErrConflict) {
245 return domain.ErrConflict251 return domain.ErrConflict
@@ -387,6 +393,8 @@ func encodeIssueState(s domain.IssueState) (storagepb.IssueState, error) {
387 switch s {393 switch s {
388 case domain.IssueStateOpen:394 case domain.IssueStateOpen:
389 return storagepb.IssueState_ISSUE_STATE_OPEN, nil395 return storagepb.IssueState_ISSUE_STATE_OPEN, nil
396+ case domain.IssueStateResolved:
397+ return storagepb.IssueState_ISSUE_STATE_RESOLVED, nil
390 case domain.IssueStateClosed:398 case domain.IssueStateClosed:
391 return storagepb.IssueState_ISSUE_STATE_CLOSED, nil399 return storagepb.IssueState_ISSUE_STATE_CLOSED, nil
392 }400 }
@@ -397,6 +405,8 @@ func decodeIssueState(s storagepb.IssueState) (domain.IssueState, error) {
397 switch s {405 switch s {
398 case storagepb.IssueState_ISSUE_STATE_OPEN:406 case storagepb.IssueState_ISSUE_STATE_OPEN:
399 return domain.IssueStateOpen, nil407 return domain.IssueStateOpen, nil
408+ case storagepb.IssueState_ISSUE_STATE_RESOLVED:
409+ return domain.IssueStateResolved, nil
400 case storagepb.IssueState_ISSUE_STATE_CLOSED:410 case storagepb.IssueState_ISSUE_STATE_CLOSED:
401 return domain.IssueStateClosed, nil411 return domain.IssueStateClosed, nil
402 case storagepb.IssueState_ISSUE_STATE_PENDING:412 case storagepb.IssueState_ISSUE_STATE_PENDING:
@@ -434,7 +444,112 @@ func decodeIssuesMode(m storagepb.IssuesMode) domain.IssuesMode {
434 // are silently dropped without preventing the associated write.444 // are silently dropped without preventing the associated write.
435 const MaxIssueEvents = 500445 const MaxIssueEvents = 500
436 446
437-// ListEvents returns issue history. Persistence is implemented in Phase A.447+// ListEvents returns the append-only issue history in sequence order.
438 func (i *Issues) ListEvents(ctx context.Context, repo domain.RepoName, number int64) ([]domain.IssueEvent, error) {448 func (i *Issues) ListEvents(ctx context.Context, repo domain.RepoName, number int64) ([]domain.IssueEvent, error) {
439- return nil, nil449+ var out []domain.IssueEvent
450+ for _, item := range i.db.Snapshot(ctx).Prefix(ctx, []byte(issueEventsPrefix(repo, number))) {
451+ if err := ctx.Err(); err != nil {
452+ return nil, err
453+ }
454+ ev, err := decodeIssueEvent(item.Value, number)
455+ if err != nil {
456+ return nil, err
457+ }
458+ out = append(out, ev)
459+ if len(out) == MaxIssueEvents {
460+ break
461+ }
462+ }
463+ return out, ctx.Err()
464+}
465+
466+func issueEventsPrefix(repo domain.RepoName, issue int64) string {
467+ return issueEventPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, issue) + "/"
468+}
469+
470+// prepareIssueEvents allocates dense per-issue sequences. RequireAbsent makes
471+// concurrent appenders conflict; callers retry the entire associated write.
472+// Scan once for a group so multiple events in one batch get distinct sequences.
473+func prepareIssueEvents(ctx context.Context, db *kv.DB, b *kv.Batch, repo domain.RepoName, issue int64, events []domain.IssueEvent) error {
474+ if len(events) == 0 {
475+ return nil
476+ }
477+ prefix := issueEventsPrefix(repo, issue)
478+ var seq int64
479+ count := 0
480+ for key := range db.Snapshot(ctx).Prefix(ctx, []byte(prefix)) {
481+ if err := ctx.Err(); err != nil {
482+ return err
483+ }
484+ n, err := strconv.ParseInt(strings.TrimPrefix(string(key), prefix), 10, 64)
485+ if err != nil {
486+ return fmt.Errorf("decode issue event key: %w", err)
487+ }
488+ seq = max(seq, n)
489+ count++
490+ }
491+ for _, ev := range events {
492+ if count >= MaxIssueEvents {
493+ break
494+ }
495+ seq++
496+ ev.ID = seq
497+ val, err := encodeIssueEvent(ev)
498+ if err != nil {
499+ return err
500+ }
501+ key := []byte(prefix + fmt.Sprintf(numberFormat, seq))
502+ b.RequireAbsent(key)
503+ b.Put(key, val)
504+ if ev.MergeRequest != 0 {
505+ b.Put([]byte(mrEventsPrefix(repo, ev.MergeRequest)+fmt.Sprintf(numberFormat, issue)+"/"+fmt.Sprintf(numberFormat, seq)), val)
506+ }
507+ count++
508+ }
509+ return ctx.Err()
510+}
511+
512+func mrEventsPrefix(repo domain.RepoName, mr int64) string {
513+ return mrEventPrefix + repo.String() + "/" + fmt.Sprintf(numberFormat, mr) + "/"
514+}
515+
516+func encodeIssueEvent(ev domain.IssueEvent) ([]byte, error) {
517+ var kind storagepb.IssueEventKind
518+ switch ev.Kind {
519+ case domain.IssueEventReferenced:
520+ kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_REFERENCED
521+ case domain.IssueEventClosed:
522+ kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_CLOSED
523+ case domain.IssueEventReopened:
524+ kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_REOPENED
525+ case domain.IssueEventResolved:
526+ kind = storagepb.IssueEventKind_ISSUE_EVENT_KIND_RESOLVED
527+ default:
528+ return nil, fmt.Errorf("%w: issue event kind %q", domain.ErrInvalid, ev.Kind)
529+ }
530+ if !utf8.ValidString(ev.Actor) {
531+ return nil, errors.New("issue event actor is not valid UTF-8")
532+ }
533+ return mustMarshal(&storagepb.IssueEvent{Seq: ev.ID, Kind: kind, MergeRequest: ev.MergeRequest, Closes: ev.Closes, Actor: ev.Actor, CreatedAt: encodeTime(ev.CreatedAt)}), nil
534+}
535+
536+func decodeIssueEvent(val []byte, issue int64) (domain.IssueEvent, error) {
537+ var pb storagepb.IssueEvent
538+ if err := proto.Unmarshal(val, &pb); err != nil {
539+ return domain.IssueEvent{}, fmt.Errorf("decode issue event: %w", err)
540+ }
541+ var kind domain.IssueEventKind
542+ switch pb.GetKind() {
543+ case storagepb.IssueEventKind_ISSUE_EVENT_KIND_REFERENCED:
544+ kind = domain.IssueEventReferenced
545+ case storagepb.IssueEventKind_ISSUE_EVENT_KIND_CLOSED:
546+ kind = domain.IssueEventClosed
547+ case storagepb.IssueEventKind_ISSUE_EVENT_KIND_REOPENED:
548+ kind = domain.IssueEventReopened
549+ case storagepb.IssueEventKind_ISSUE_EVENT_KIND_RESOLVED:
550+ kind = domain.IssueEventResolved
551+ default:
552+ return domain.IssueEvent{}, fmt.Errorf("unknown issue event kind %d", pb.GetKind())
553+ }
554+ return domain.IssueEvent{ID: pb.GetSeq(), Issue: issue, Kind: kind, MergeRequest: pb.GetMergeRequest(), Closes: pb.GetCloses(), Actor: pb.GetActor(), CreatedAt: decodeTime(pb.GetCreatedAt())}, nil
440 }555 }
M internal/adapters/kvstore/issues_test.go +1 -1
@@ -364,7 +364,7 @@ func TestSchemaNewerThanThisBuildStopsIssuesData(t *testing.T) {
364364 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: kvstore.SchemaVersion + 1}))
365365 _, err := kvstore.Open(ctx, dir)
366366 require.ErrorIs(t, err, kvstore.ErrSchemaTooNew)
367- assert.Equal(t, uint32(7), uint32(kvstore.SchemaVersion))
367+ assert.Equal(t, uint32(8), uint32(kvstore.SchemaVersion))
368368 }
369369
370370 // A reopen racing the delete of the repository either lands before the delete
@@ -364,7 +364,7 @@ func TestSchemaNewerThanThisBuildStopsIssuesData(t *testing.T) {
364 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: kvstore.SchemaVersion + 1}))364 putRaw(t, dir, "meta/schema", mustProto(t, &storagepb.SchemaVersion{Version: kvstore.SchemaVersion + 1}))
365 _, err := kvstore.Open(ctx, dir)365 _, err := kvstore.Open(ctx, dir)
366 require.ErrorIs(t, err, kvstore.ErrSchemaTooNew)366 require.ErrorIs(t, err, kvstore.ErrSchemaTooNew)
367- assert.Equal(t, uint32(7), uint32(kvstore.SchemaVersion))367+ assert.Equal(t, uint32(8), uint32(kvstore.SchemaVersion))
368 }368 }
369 369
370 // A reopen racing the delete of the repository either lands before the delete370 // A reopen racing the delete of the repository either lands before the delete
M internal/adapters/kvstore/store.go +34 -1
@@ -44,7 +44,9 @@ import (
4444 // adds releases and release:write; older builds cannot clean up their metadata
4545 // or preserve the owner rights of upgraded tokens (see upgradeReleaseTokens).
4646 // Version 7 adds the latest-release index; older writers would leave it stale.
47-const SchemaVersion = 7
47+// Version 8 adds resolved issues and history (isev/, mrev/), and rewrites
48+// merge-closed issues as resolved; older builds cannot decode state 4.
49+const SchemaVersion = 8
4850
4951 const schemaKey = "meta/schema"
5052
@@ -172,6 +174,11 @@ func (s *Store) checkSchema(ctx context.Context) error {
172174 return err
173175 }
174176 }
177+ if target == SchemaVersion {
178+ if err := s.upgradeResolvedIssues(ctx, &b); err != nil {
179+ return err
180+ }
181+ }
175182 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: target}))
176183 _, err := s.db.Apply(ctx, &b)
177184 if errors.Is(err, kv.ErrConflict) {
@@ -274,3 +281,29 @@ func (s *Store) upgradeTokenScopes(ctx context.Context, b *kv.Batch, required, a
274281 }
275282 return nil
276283 }
284+
285+// upgradeResolvedIssues runs in the schema batch, without retroactive events or
286+// timestamp/index changes. Typical records are <1 KiB; the 16 MiB batch limit
287+// accommodates over 10k issues, consistent with the unchunked token upgrades.
288+func (s *Store) upgradeResolvedIssues(ctx context.Context, b *kv.Batch) error {
289+ count := 0
290+ for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(issuePrefix)) {
291+ if err := ctx.Err(); err != nil {
292+ return err
293+ }
294+ var pb storagepb.Issue
295+ if err := proto.Unmarshal(item.Value, &pb); err != nil {
296+ return fmt.Errorf("upgrade issue %s: %w", key, err)
297+ }
298+ if pb.GetState() != storagepb.IssueState_ISSUE_STATE_CLOSED || pb.GetClosedByMergeRequest() == 0 {
299+ continue
300+ }
301+ pb.State = storagepb.IssueState_ISSUE_STATE_RESOLVED
302+ b.RequireVersion(key, item.Version)
303+ b.Put(key, mustMarshal(&pb))
304+ count++
305+ }
306+ log := zerowrap.FromCtx(ctx)
307+ log.Info().Int("count", count).Msg("merge-closed issues upgraded to resolved")
308+ return ctx.Err()
309+}
@@ -44,7 +44,9 @@ import (
44 // adds releases and release:write; older builds cannot clean up their metadata44 // adds releases and release:write; older builds cannot clean up their metadata
45 // or preserve the owner rights of upgraded tokens (see upgradeReleaseTokens).45 // or preserve the owner rights of upgraded tokens (see upgradeReleaseTokens).
46 // Version 7 adds the latest-release index; older writers would leave it stale.46 // Version 7 adds the latest-release index; older writers would leave it stale.
47-const SchemaVersion = 747+// Version 8 adds resolved issues and history (isev/, mrev/), and rewrites
48+// merge-closed issues as resolved; older builds cannot decode state 4.
49+const SchemaVersion = 8
48 50
49 const schemaKey = "meta/schema"51 const schemaKey = "meta/schema"
50 52
@@ -172,6 +174,11 @@ func (s *Store) checkSchema(ctx context.Context) error {
172 return err174 return err
173 }175 }
174 }176 }
177+ if target == SchemaVersion {
178+ if err := s.upgradeResolvedIssues(ctx, &b); err != nil {
179+ return err
180+ }
181+ }
175 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: target}))182 b.Put(key, mustMarshal(&storagepb.SchemaVersion{Version: target}))
176 _, err := s.db.Apply(ctx, &b)183 _, err := s.db.Apply(ctx, &b)
177 if errors.Is(err, kv.ErrConflict) {184 if errors.Is(err, kv.ErrConflict) {
@@ -274,3 +281,29 @@ func (s *Store) upgradeTokenScopes(ctx context.Context, b *kv.Batch, required, a
274 }281 }
275 return nil282 return nil
276 }283 }
284+
285+// upgradeResolvedIssues runs in the schema batch, without retroactive events or
286+// timestamp/index changes. Typical records are <1 KiB; the 16 MiB batch limit
287+// accommodates over 10k issues, consistent with the unchunked token upgrades.
288+func (s *Store) upgradeResolvedIssues(ctx context.Context, b *kv.Batch) error {
289+ count := 0
290+ for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(issuePrefix)) {
291+ if err := ctx.Err(); err != nil {
292+ return err
293+ }
294+ var pb storagepb.Issue
295+ if err := proto.Unmarshal(item.Value, &pb); err != nil {
296+ return fmt.Errorf("upgrade issue %s: %w", key, err)
297+ }
298+ if pb.GetState() != storagepb.IssueState_ISSUE_STATE_CLOSED || pb.GetClosedByMergeRequest() == 0 {
299+ continue
300+ }
301+ pb.State = storagepb.IssueState_ISSUE_STATE_RESOLVED
302+ b.RequireVersion(key, item.Version)
303+ b.Put(key, mustMarshal(&pb))
304+ count++
305+ }
306+ log := zerowrap.FromCtx(ctx)
307+ log.Info().Int("count", count).Msg("merge-closed issues upgraded to resolved")
308+ return ctx.Err()
309+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 4

Tags 45

↑↓ moveenter openesc close