-
1
-
2
-
3
-
4
-
5
-
6
-
7
-
8
-
9
-
10
-
11
-
12
-
13
-
14
-
15
-
16
-
17
-
18
-
19
-
20
-
21
-
22
-
23
-
24
-
25
-
26
-
27
-
28
-
29
-
30
-
31
-
32
-
33
-
34
-
35
-
36
-
37
-
38
-
39
-
40
-
41
-
42
-
43
-
44
-
45
-
46
-
47
-
48
-
49
-
50
-
51
-
52
-
53
-
54
-
55
-
56
-
57
-
58
-
59
-
60
-
61
-
62
-
63
-
64
-
65
-
66
-
67
-
68
-
69
-
70
-
71
-
72
-
73
// SPDX-FileCopyrightText: 2025 Shota FUJI <pockawoooh@gmail.com>
// SPDX-License-Identifier: AGPL-3.0-only
package snapshot
import (
"database/sql"
"google.golang.org/protobuf/proto"
)
// Zig に戻りたい...。
// https://konradreiche.com/blog/a-generic-protobuf-reader-with-go-type-parameters/
type ProtoMessage[T any] interface {
proto.Message
*T
}
func GetLatest[T any, P ProtoMessage[T]](tx *sql.Tx) (P, uint64, error) {
var ret P = new(T)
stmt, err := tx.Prepare(`
SELECT event_seq, payload FROM snapshots
LEFT JOIN projections ON snapshots.projection_id = projections.id
WHERE projection_name = ?
ORDER BY event_seq DESC
LIMIT 1
`)
if err != nil {
return nil, 0, err
}
row := stmt.QueryRow(ret.ProtoReflect().Descriptor().FullName())
var seq uint64
var payload []byte
err = row.Scan(&seq, &payload)
if err == sql.ErrNoRows {
return nil, 0, nil
}
if err != nil {
return nil, 0, err
}
if err := proto.Unmarshal(payload, ret); err != nil {
return nil, 0, err
}
return ret, seq, nil
}
func Save(tx *sql.Tx, msg proto.Message, seq uint64) error {
name := msg.ProtoReflect().Descriptor().FullName()
payload, err := proto.Marshal(msg)
if err != nil {
return err
}
_, err = tx.Exec(`
INSERT INTO projections (projection_name)
SELECT ?1 WHERE NOT EXISTS (SELECT id FROM projections WHERE projection_name = ?1)
`, name)
if err != nil {
return err
}
_, err = tx.Exec(`
INSERT INTO snapshots (event_seq, projection_id, payload)
SELECT ?, id, ? FROM projections WHERE projection_name = ? LIMIT 1
`, seq, payload, name)
return err
}