1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package kvstore
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "strings"
10 "time"
11
12 kv "bnema.dev/ermite-kv"
13 "google.golang.org/protobuf/proto"
14
15 storagepb "bnema.dev/ermite/gen/ermite/storage"
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20var _ ports.MirrorStore = (*Store)(nil)
21var _ ports.MirrorKeyStore = (*Store)(nil)
22
23const mirrorConfigKey = "mirror/config"
24const mirrorRepoPrefix = "mirror/repo/"
25const mirrorStatePrefix = "mirror/state/"
26const mirrorSnapshotPrefix = "mirror/snapshot/"
27
28func mirrorKeys(ctx context.Context, snap *kv.Snapshot, repo domain.RepoName) [][]byte {
29 var keys [][]byte
30 for _, prefix := range []string{mirrorRepoPrefix, mirrorStatePrefix, mirrorSnapshotPrefix} {
31 key := []byte(prefix + repo.String())
32 if _, ok := snap.Get(ctx, key); ok {
33 keys = append(keys, key)
34 }
35 }
36 return keys
37}
38
39func (s *Store) readMirror(ctx context.Context, key string, pb proto.Message) error {
40 item, ok := s.db.Get(ctx, []byte(key))
41 if !ok {
42 return domain.ErrNotFound
43 }
44 if err := proto.Unmarshal(item.Value, pb); err != nil {
45 return fmt.Errorf("decode mirror record: %w", err)
46 }
47 return nil
48}
49
50func (s *Store) writeMirror(ctx context.Context, key string, repo domain.RepoName, pb proto.Message) error {
51 val, err := marshalMirror(pb)
52 if err != nil {
53 return err
54 }
55 for range updateAttempts {
56 var b kv.Batch
57 if key != mirrorConfigKey {
58 if err := guardRepoKey(ctx, s.db, &b, repo); err != nil {
59 return err
60 }
61 }
62 b.Put([]byte(key), val)
63 _, err := s.db.Apply(ctx, &b)
64 if errors.Is(err, kv.ErrConflict) {
65 continue
66 }
67 if err != nil {
68 return fmt.Errorf("store mirror record: %w", err)
69 }
70 return nil
71 }
72 return domain.ErrConflict
73}
74
75// marshalMirror refuses what protobuf cannot encode: a status detail comes from remote output
76// and may hold invalid UTF-8, which would make mustMarshal panic.
77func marshalMirror(pb proto.Message) ([]byte, error) {
78 val, err := proto.MarshalOptions{Deterministic: true}.Marshal(pb)
79 if err != nil {
80 return nil, fmt.Errorf("%w: mirror record is not valid UTF-8", domain.ErrInvalid)
81 }
82 return val, nil
83}
84
85func (s *Store) GetGlobal(ctx context.Context) (domain.MirrorGlobalConfig, error) {
86 var pb storagepb.MirrorGlobal
87 err := s.readMirror(ctx, mirrorConfigKey, &pb)
88 if errors.Is(err, domain.ErrNotFound) {
89 err = nil
90 }
91 return domain.MirrorGlobalConfig{Enabled: pb.Enabled, URLTemplate: pb.UrlTemplate, Cron: pb.Cron, AllBranches: pb.AllBranches, UpdatedAt: decodeTime(pb.UpdatedAt)}, err
92}
93func (s *Store) PutGlobal(ctx context.Context, g domain.MirrorGlobalConfig) error {
94 return s.writeMirror(ctx, mirrorConfigKey, domain.RepoName{}, &storagepb.MirrorGlobal{Enabled: g.Enabled, UrlTemplate: g.URLTemplate, Cron: g.Cron, AllBranches: g.AllBranches, UpdatedAt: encodeTime(g.UpdatedAt)})
95}
96func decodeMirrorRepo(repo domain.RepoName, pb *storagepb.MirrorRepo) domain.MirrorRepoConfig {
97 return domain.MirrorRepoConfig{Repo: repo, URL: pb.Url, Cron: pb.Cron, AllBranches: pb.AllBranches, Excluded: pb.Excluded, UpdatedAt: decodeTime(pb.UpdatedAt)}
98}
99func (s *Store) GetRepoConfig(ctx context.Context, repo domain.RepoName) (domain.MirrorRepoConfig, error) {
100 var pb storagepb.MirrorRepo
101 err := s.readMirror(ctx, mirrorRepoPrefix+repo.String(), &pb)
102 return decodeMirrorRepo(repo, &pb), err
103}
104func (s *Store) PutRepoConfig(ctx context.Context, c domain.MirrorRepoConfig) error {
105 return s.writeMirror(ctx, mirrorRepoPrefix+c.Repo.String(), c.Repo, &storagepb.MirrorRepo{Url: c.URL, Cron: c.Cron, AllBranches: c.AllBranches, Excluded: c.Excluded, UpdatedAt: encodeTime(c.UpdatedAt)})
106}
107func (s *Store) DeleteRepoConfig(ctx context.Context, repo domain.RepoName) error {
108 var b kv.Batch
109 b.Delete([]byte(mirrorRepoPrefix + repo.String()))
110 _, err := s.db.Apply(ctx, &b)
111 if err != nil {
112 return fmt.Errorf("delete mirror config: %w", err)
113 }
114 return nil
115}
116func (s *Store) ListRepoConfigs(ctx context.Context) ([]domain.MirrorRepoConfig, error) {
117 var out []domain.MirrorRepoConfig
118 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mirrorRepoPrefix)) {
119 repo, err := domain.ParseRepoName(strings.TrimPrefix(string(key), mirrorRepoPrefix))
120 if err != nil {
121 return nil, err
122 }
123 var pb storagepb.MirrorRepo
124 if err := proto.Unmarshal(item.Value, &pb); err != nil {
125 return nil, fmt.Errorf("decode mirror config: %w", err)
126 }
127 out = append(out, decodeMirrorRepo(repo, &pb))
128 }
129 return out, nil
130}
131func decodeMirrorStatus(repo domain.RepoName, pb *storagepb.MirrorState) domain.MirrorStatus {
132 return domain.MirrorStatus{Repo: repo, State: domain.MirrorState(pb.State), Class: domain.MirrorFailureClass(pb.Class), Reason: pb.Reason, Detail: pb.Detail, ConsecutiveFailures: int(pb.ConsecutiveFailures), LastAttemptAt: decodeTime(pb.LastAttemptAt), LastSuccessAt: decodeTime(pb.LastSuccessAt), NextRetryAt: decodeTime(pb.NextRetryAt), PausedAt: decodeTime(pb.PausedAt), UpdatedAt: decodeTime(pb.UpdatedAt)}
133}
134func (s *Store) GetStatus(ctx context.Context, repo domain.RepoName) (domain.MirrorStatus, error) {
135 var pb storagepb.MirrorState
136 err := s.readMirror(ctx, mirrorStatePrefix+repo.String(), &pb)
137 return decodeMirrorStatus(repo, &pb), err
138}
139func (s *Store) PutStatus(ctx context.Context, c domain.MirrorStatus) error {
140 // Detail is the tail of a remote's stderr, whose bytes are not ours: repair rather than lose
141 // the status (which would stall the incident).
142 c.Detail = strings.ToValidUTF8(c.Detail, "?")
143 c.Reason = strings.ToValidUTF8(c.Reason, "?")
144 return s.writeMirror(ctx, mirrorStatePrefix+c.Repo.String(), c.Repo, &storagepb.MirrorState{State: string(c.State), Class: string(c.Class), Reason: c.Reason, Detail: c.Detail, ConsecutiveFailures: int32(c.ConsecutiveFailures), LastAttemptAt: encodeTime(c.LastAttemptAt), LastSuccessAt: encodeTime(c.LastSuccessAt), NextRetryAt: encodeTime(c.NextRetryAt), PausedAt: encodeTime(c.PausedAt), UpdatedAt: encodeTime(c.UpdatedAt)})
145}
146func (s *Store) ListStatus(ctx context.Context) ([]domain.MirrorStatus, error) {
147 var out []domain.MirrorStatus
148 for key, item := range s.db.Snapshot(ctx).Prefix(ctx, []byte(mirrorStatePrefix)) {
149 repo, err := domain.ParseRepoName(strings.TrimPrefix(string(key), mirrorStatePrefix))
150 if err != nil {
151 return nil, err
152 }
153 var pb storagepb.MirrorState
154 if err := proto.Unmarshal(item.Value, &pb); err != nil {
155 return nil, fmt.Errorf("decode mirror state: %w", err)
156 }
157 out = append(out, decodeMirrorStatus(repo, &pb))
158 }
159 return out, nil
160}
161func (s *Store) GetSnapshot(ctx context.Context, repo domain.RepoName) (domain.MirrorSnapshot, error) {
162 var pb storagepb.MirrorSnapshot
163 err := s.readMirror(ctx, mirrorSnapshotPrefix+repo.String(), &pb)
164 out := domain.MirrorSnapshot{URL: pb.Url, AllBranches: pb.AllBranches, At: decodeTime(pb.At)}
165 for _, ref := range pb.Refs {
166 id, parseErr := domain.ParseObjectID(ref.Id)
167 if parseErr != nil {
168 return domain.MirrorSnapshot{}, fmt.Errorf("decode mirror snapshot ref: %w", parseErr)
169 }
170 out.Refs = append(out.Refs, domain.MirrorRef{Name: ref.Name, ID: id})
171 }
172 return out, err
173}
174func (s *Store) PutSnapshot(ctx context.Context, repo domain.RepoName, c domain.MirrorSnapshot) error {
175 pb := &storagepb.MirrorSnapshot{Url: c.URL, AllBranches: c.AllBranches, At: encodeTime(c.At)}
176 for _, ref := range c.Refs {
177 pb.Refs = append(pb.Refs, &storagepb.MirrorSnapshotRef{Name: ref.Name, Id: ref.ID.String()})
178 }
179 return s.writeMirror(ctx, mirrorSnapshotPrefix+repo.String(), repo, pb)
180}
181func (s *Store) GetMirrorKey(ctx context.Context) ([]byte, time.Time, error) {
182 var pb storagepb.MirrorKey
183 err := s.readMirror(ctx, domain.MirrorKeyPath, &pb)
184 return pb.Sealed, decodeTime(pb.CreatedAt), err
185}
186func (s *Store) PutMirrorKey(ctx context.Context, sealed []byte, at time.Time, ifAbsent bool) error {
187 var b kv.Batch
188 key := []byte(domain.MirrorKeyPath)
189 if ifAbsent {
190 b.RequireAbsent(key)
191 }
192 b.Put(key, mustMarshal(&storagepb.MirrorKey{Sealed: sealed, CreatedAt: encodeTime(at)})) // bytes and a timestamp: always encodable
193 _, err := s.db.Apply(ctx, &b)
194 if ifAbsent && errors.Is(err, kv.ErrConflict) {
195 return domain.ErrAlreadyExists
196 }
197 if err != nil {
198 return fmt.Errorf("store mirror key: %w", err)
199 }
200 return nil
201}