go.temporal.io/server/chasm/visibility.go
409 LOC · 170 covered · 239 uncovered · 55 ranges · 1230 concepts · 34 introducers · 586 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package chasm
2
3
import (
4
"context"
5
"fmt"
6
"maps"
7
"strings"
8
9
commonpb "go.temporal.io/api/common/v1"
10
enumspb "go.temporal.io/api/enums/v1"
11
"go.temporal.io/api/serviceerror"
12
persistencespb "go.temporal.io/server/api/persistence/v1"
13
"go.temporal.io/server/common/payload"
14
"go.temporal.io/server/common/searchattribute/sadefs"
15
"google.golang.org/protobuf/proto"
16
)
17
18
const (
19
UserMemoKey = "__user__"
20
ChasmMemoKey = "__chasm__"
21
22
visibilityComponentType = "core.vis"
23
visibilityTaskType = "core.visTask"
24
)
25
26
var (
27
visibilityComponentTypeID = GenerateTypeID(visibilityComponentType)
28
visibilityTaskTypeID = GenerateTypeID(visibilityTaskType)
29
)
30
31
// VisibilitySearchAttributesProvider if implemented by the root Component,
32
// allows the CHASM framework to automatically determine, at the end of
33
// a transaction, if a visibility task needs to be generated to update the
34
// visibility record with the returned search attributes.
35
type VisibilitySearchAttributesProvider interface {
36
SearchAttributes(Context) []SearchAttributeKeyValue
37
}
38
39
// VisibilityMemoProvider if implemented by the root Component,
40
// allows the CHASM framework to automatically determine, at the end of
41
// a transaction, if a visibility task needs to be generated to update the
42
// visibility record with the returned memo.
43
type VisibilityMemoProvider interface {
44
Memo(Context) proto.Message
45
}
46
47
// VisibilitySearchAttributesMapper is a mapper for CHASM search attributes.
48
type VisibilitySearchAttributesMapper struct {
49
// map from CHASM and predefined search attribute aliases to field names.
50
aliasToField map[string]string
51
fieldToAlias map[string]string
52
saTypeMap map[string]enumspb.IndexedValueType
53
54
// systemAliasToField maps a CHASM search attribute alias to a system field
55
// (e.g. "ScheduleId" -> "WorkflowId"). Used to resolve system search attribute aliases,
56
// including the businessID alias configured via WithBusinessIDAlias.
57
systemAliasToField map[string]string
58
59
// overriddenSystemFields records system search attribute fields (e.g. ExecutionTime, TaskQueue)
60
// this archetype overrides with its own value, stored in the dedicated system column. Value is
61
// the field's indexed value type.
62
overriddenSystemFields map[string]enumspb.IndexedValueType
63
}
64
65
// newVisibilitySearchAttributesMapper returns a mapper with all maps initialized.
66
>
func newVisibilitySearchAttributesMapper() *VisibilitySearchAttributesMapper {
visibility.go ×1
67
>
return &VisibilitySearchAttributesMapper{
68
>
aliasToField: make(map[string]string),
69
>
fieldToAlias: make(map[string]string),
70
>
saTypeMap: make(map[string]enumspb.IndexedValueType),
71
>
systemAliasToField: make(map[string]string),
72
>
overriddenSystemFields: make(map[string]enumspb.IndexedValueType),
73
>
}
74
>
}
75
76
// Alias returns the alias for a given field.
77
func (v *VisibilitySearchAttributesMapper) Alias(field string) (string, error) {
78
if v == nil {
79
return "", serviceerror.NewInvalidArgument("visibility search attributes mapper not defined")
80
}
81
alias, ok := v.fieldToAlias[field]
82
if !ok {
83
return "", serviceerror.NewInvalidArgumentf(
84
"visibility search attributes mapper has no registered field %q",
85
field,
86
)
87
}
88
return alias, nil
89
}
90
91
// Field returns the field for a given alias.
92
>
func (v *VisibilitySearchAttributesMapper) Field(alias string) (string, error) {
visibility.go ×2
93
>
if v == nil {
94
return "", serviceerror.NewInvalidArgument("visibility search attributes mapper not defined")
95
}
98
>
}
100
return field, nil
101
}
102
>
return "", serviceerror.NewInvalidArgument(fmt.Sprintf("visibility search attributes mapper has no registered alias %q", alias))
visibility.go ×3
103
}
104
105
// resolveSystemAlias resolves a system search attribute alias to its field name.
106
// It handles the `Temporal` prefix variations (e.g., "ScheduleId" and "TemporalScheduleId").
107
>
func (v *VisibilitySearchAttributesMapper) resolveSystemAlias(alias string) (string, bool) {
visibility.go ×3
108
>
if v.systemAliasToField == nil {
110
>
}
112
return field, true
113
}
114
// Try without the `Temporal` prefix.
116
withoutPrefix := alias[len(sadefs.ReservedPrefix):]
117
if field, ok := v.systemAliasToField[withoutPrefix]; ok {
118
return field, true
119
}
121
>
// Try with the `Temporal` prefix.
122
>
withPrefix := sadefs.ReservedPrefix + alias
123
>
if field, ok := v.systemAliasToField[withPrefix]; ok {
124
return field, true
125
}
126
}
128
}
129
130
// SATypeMap returns the type map for the CHASM search attributes.
131
>
func (v *VisibilitySearchAttributesMapper) SATypeMap() map[string]enumspb.IndexedValueType {
visibility.go ×1
132
>
if v == nil {
134
>
}
136
}
137
138
// IsSystemOverride returns true if this archetype overrides the given system search attribute
139
// field with its own value (written to the dedicated system column).
140
>
func (v *VisibilitySearchAttributesMapper) IsSystemOverride(field string) bool {
visibility.go ×2
141
>
if v == nil {
142
return false
143
}
145
>
return ok
146
}
147
148
// OverriddenSystemFields returns the system search attribute fields this archetype overrides,
149
// keyed by field name with the field's indexed value type as the value.
150
>
func (v *VisibilitySearchAttributesMapper) OverriddenSystemFields() map[string]enumspb.IndexedValueType {
visibility.go ×1
151
>
if v == nil {
153
>
}
155
}
156
157
// ValueType returns the type of a CHASM search attribute field.
158
// Returns an error if the field is not found in the type map.
159
>
func (v *VisibilitySearchAttributesMapper) ValueType(fieldName string) (enumspb.IndexedValueType, error) {
visibility.go ×3
160
>
if v == nil {
161
return enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED, serviceerror.NewInvalidArgument("visibility search attributes mapper not defined")
162
}
164
>
if !ok {
165
>
return enumspb.INDEXED_VALUE_TYPE_UNSPECIFIED, serviceerror.NewInvalidArgumentf("visibility search attributes mapper has no registered field %q", fieldName)
visibility.go ×1
166
>
}
168
}
169
170
type Visibility struct {
171
UnimplementedComponent
172
173
Data *persistencespb.ChasmVisibilityData
174
175
// Do NOT access those fields directly.
176
// Use the provided getters and setters instead.
177
SA Field[*commonpb.SearchAttributes]
178
Memo Field[*commonpb.Memo]
179
}
180
181
func NewVisibility(
182
mutableContext MutableContext,
184
>
visibility := &Visibility{
185
>
Data: &persistencespb.ChasmVisibilityData{
186
>
TransitionCount: 0,
187
>
},
188
>
}
189
>
190
>
visibility.generateTask(mutableContext)
191
>
return visibility
192
>
}
193
194
func NewVisibilityWithData(
195
mutableContext MutableContext,
196
customSearchAttributes map[string]*commonpb.Payload,
197
customMemo map[string]*commonpb.Payload,
199
>
visibility := &Visibility{
200
>
Data: &persistencespb.ChasmVisibilityData{
201
>
TransitionCount: 0,
202
>
},
203
>
}
204
>
205
>
// Filter out nil/empty payload values for search attributes.
206
>
filteredSA := payload.MergeMapOfPayload(nil, customSearchAttributes)
207
>
if len(filteredSA) != 0 {
209
>
mutableContext,
210
>
&commonpb.SearchAttributes{IndexedFields: filteredSA},
211
>
)
212
>
}
213
214
// Filter out nil/empty payload values for memo.
216
>
if len(filteredMemo) != 0 {
218
>
mutableContext,
219
>
&commonpb.Memo{Fields: filteredMemo},
220
>
)
221
>
}
222
224
>
return visibility
225
}
226
228
>
return LifecycleStateRunning
229
>
}
230
231
// CustomSearchAttributes returns the stored custom search attribute fields.
232
// Nil is returned if there are none.
233
//
234
// Returned map is a shallow copy: callers may add, delete, or reassign keys without
235
// affecting the stored data, but the *commonpb.Payload values are shared.
236
func (v *Visibility) CustomSearchAttributes(
237
chasmContext Context,
239
>
sa, _ := v.SA.TryGet(chasmContext)
240
>
// nil check handled by the proto getter.
241
>
return maps.Clone(sa.GetIndexedFields())
242
>
}
243
244
// MergeCustomSearchAttributes merges the provided custom search attribute fields into the existing ones.
245
// - If a key in `customSearchAttributes` already exists,
246
// the value in `customSearchAttributes` replaces the existing value.
247
// - If a key in `customSearchAttributes` has nil or empty slice payload value,
248
// the key is deleted from the existing search attributes if it exists.
249
// If all search attributes are removed, the underlying search attributes node is deleted.
250
// - If `customSearchAttributes` is empty, this is a no-op.
251
func (v *Visibility) MergeCustomSearchAttributes(
252
mutableContext MutableContext,
253
customSearchAttributes map[string]*commonpb.Payload,
255
>
if len(customSearchAttributes) == 0 {
257
>
}
258
260
>
if !ok {
261
>
currentSA = &commonpb.SearchAttributes{}
262
>
v.SA = NewDataField(mutableContext, currentSA)
263
>
}
264
266
>
currentSA.GetIndexedFields(),
267
>
customSearchAttributes,
268
>
)
269
>
if len(currentSA.IndexedFields) == 0 {
271
>
}
272
274
}
275
276
// ReplaceCustomSearchAttributes replaces the existing custom search attribute fields with the provided ones.
277
// Nil/empty payload values are filtered.
278
// If `customSearchAttributes` is empty or all values are nil after filtering, the underlying search attributes node is deleted.
279
func (v *Visibility) ReplaceCustomSearchAttributes(
280
mutableContext MutableContext,
281
customSearchAttributes map[string]*commonpb.Payload,
283
>
// Filter out nil/empty payload values.
284
>
filteredSA := payload.MergeMapOfPayload(nil, customSearchAttributes)
285
>
286
>
if len(filteredSA) == 0 {
287
>
_, ok := v.SA.TryGet(mutableContext)
288
>
if !ok {
289
// Already empty, no-op
290
return
291
}
292
294
>
} else {
295
>
v.SA = NewDataField(
296
>
mutableContext,
297
>
&commonpb.SearchAttributes{IndexedFields: filteredSA},
298
>
)
299
>
}
300
302
}
303
304
// CustomMemo returns the stored custom memo fields.
305
// Nil is returned if there are none.
306
//
307
// Returned map is a shallow copy: callers may add, delete, or reassign keys without
308
// affecting the stored data, but the *commonpb.Payload values are shared.
309
func (v *Visibility) CustomMemo(
310
chasmContext Context,
312
>
memo, _ := v.Memo.TryGet(chasmContext)
313
>
// nil check handled by the proto getter.
314
>
return maps.Clone(memo.GetFields())
315
>
}
316
317
// MergeCustomMemo merges the provided custom memo fields into the existing ones.
318
// - If a key in `customMemo` already exists,
319
// the value in `customMemo` replaces the existing value.
320
// - If a key in `customMemo` has nil or empty slice payload value,
321
// the key is deleted from the existing memo if it exists.
322
// If all memo fields are removed, the underlying memo node is deleted.
323
// - If `customMemo` is empty, this is a no-op.
324
func (v *Visibility) MergeCustomMemo(
325
mutableContext MutableContext,
326
customMemo map[string]*commonpb.Payload,
328
>
if len(customMemo) == 0 {
330
>
}
331
333
>
if !ok {
334
>
currentMemo = &commonpb.Memo{}
335
>
v.Memo = NewDataField(mutableContext, currentMemo)
336
>
}
337
339
>
currentMemo.GetFields(),
340
>
customMemo,
341
>
)
342
>
if len(currentMemo.Fields) == 0 {
344
>
}
346
}
347
348
// ReplaceCustomMemo replaces the existing custom memo fields with the provided ones.
349
// If `customMemo` is empty, the underlying memo node is deleted.
350
func (v *Visibility) ReplaceCustomMemo(
351
mutableContext MutableContext,
352
customMemo map[string]*commonpb.Payload,
354
>
// Filter out nil/empty payload values for memo.
355
>
filteredMemo := payload.MergeMapOfPayload(nil, customMemo)
356
>
357
>
if len(filteredMemo) == 0 {
359
>
if !ok {
360
// Already empty, no-op
361
return
362
}
363
366
>
v.Memo = NewDataField(
367
>
mutableContext,
368
>
&commonpb.Memo{Fields: filteredMemo},
369
>
)
370
>
}
371
373
}
374
375
func (v *Visibility) generateTask(
376
mutableContext MutableContext,
378
>
v.Data.TransitionCount++
379
>
mutableContext.AddTask(
380
>
v,
381
>
TaskAttributes{},
382
>
&persistencespb.ChasmVisibilityTaskData{TransitionCount: v.Data.TransitionCount},
383
>
)
384
>
}
385
386
type visibilityTaskHandler struct {
387
SideEffectTaskHandlerBase[*persistencespb.ChasmVisibilityTaskData]
388
}
389
390
var defaultVisibilityTaskHandler = &visibilityTaskHandler{}
391
392
func (v *visibilityTaskHandler) Validate(
393
_ Context,
394
component *Visibility,
395
_ TaskInvocation,
396
task *persistencespb.ChasmVisibilityTaskData,
398
>
return task.TransitionCount == component.Data.TransitionCount, nil
399
>
}
400
401
func (v *visibilityTaskHandler) Execute(
402
_ context.Context,
403
_ ComponentRef,
404
_ TaskAttributes,
405
_ *persistencespb.ChasmVisibilityTaskData,
406
) error {
407
//nolint:forbidigo
408
panic("chasm visibilityTaskHandler should not be called directly")
409
}