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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
package eventstream
import (
"context"
"encoding/json"
"errors"
"fmt"
"strconv"
"strings"
"github.com/johnandersen777/socialweb-computer-kcp/api/v1alpha1"
)
type Match struct {
FieldPath []string
Value string
}
type ValueFunc func(ctx context.Context, from v1alpha1.AtprotoValueFrom) (map[string]any, error)
func ResolveMatches(ctx context.Context, spec []v1alpha1.AtprotoRecordMatch, value ValueFunc) ([]Match, error) {
if len(spec) == 0 {
return nil, nil
}
out := make([]Match, 0, len(spec))
for _, match := range spec {
path := splitFieldPath(match.FieldPath)
if len(path) == 0 {
return nil, errors.New("eventstream: a match needs a fieldPath")
}
resolved, err := matchValue(ctx, match, value)
if err != nil {
return nil, err
}
out = append(out, Match{FieldPath: path, Value: resolved})
}
return out, nil
}
func matchValue(ctx context.Context, match v1alpha1.AtprotoRecordMatch, value ValueFunc) (string, error) {
if match.ValueFrom == nil {
return match.Value, nil
}
from := *match.ValueFrom
if strings.TrimSpace(from.Name) == "" {
return "", errors.New("eventstream: a match valueFrom needs a name")
}
path := splitFieldPath(from.FieldPath)
if len(path) == 0 {
return "", errors.New("eventstream: a match valueFrom needs a fieldPath")
}
if value == nil {
return "", fmt.Errorf("eventstream: match valueFrom %s %s has no reader", from.Kind, from.Name)
}
object, err := value(ctx, from)
if err != nil {
return "", fmt.Errorf("eventstream: match valueFrom %s %s: %w", from.Kind, from.Name, err)
}
resolved, ok := recordFieldString(object, path)
if !ok {
return "", fmt.Errorf("eventstream: field %s is missing", strings.Join(path, "."))
}
return resolved, nil
}
func splitFieldPath(path string) []string {
trimmed := strings.TrimSpace(path)
if trimmed == "" {
return nil
}
parts := strings.Split(trimmed, ".")
out := make([]string, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(part)
if part != "" {
out = append(out, part)
}
}
return out
}
func scalarString(value any) string {
switch typed := value.(type) {
case nil:
return ""
case string:
return typed
case bool:
return strconv.FormatBool(typed)
case int64:
return strconv.FormatInt(typed, 10)
case int:
return strconv.Itoa(typed)
case float64:
if typed == float64(int64(typed)) {
return strconv.FormatInt(int64(typed), 10)
}
return strconv.FormatFloat(typed, 'f', -1, 64)
case json.Number:
return typed.String()
default:
raw, err := json.Marshal(typed)
if err != nil {
return fmt.Sprint(typed)
}
return string(raw)
}
}
func recordFieldString(value any, path []string) (string, bool) {
current := value
for _, part := range path {
object, ok := current.(map[string]any)
if !ok {
return "", false
}
current, ok = object[part]
if !ok {
return "", false
}
}
return scalarString(current), true
}
func MatchesRecord(value json.RawMessage, matches []Match) bool {
var decoded any
if err := json.Unmarshal(value, &decoded); err != nil {
return false
}
for _, match := range matches {
actual, ok := recordFieldString(decoded, match.FieldPath)
if !ok || actual != match.Value {
return false
}
}
return true
}
