Watch returns a channel that emits a stream of changes to the relationship tuples in the database.
(ctx context.Context, tenantID, snap string)
| 44 | |
| 45 | // Watch returns a channel that emits a stream of changes to the relationship tuples in the database. |
| 46 | func (w *Watch) Watch(ctx context.Context, tenantID, snap string) (<-chan *base.DataChanges, <-chan error) { |
| 47 | // Create channels for changes and errors. |
| 48 | changes := make(chan *base.DataChanges, w.database.GetWatchBufferSize()) |
| 49 | errs := make(chan error, 1) |
| 50 | |
| 51 | var sleep *time.Timer |
| 52 | const maxSleepDuration = 2 * time.Second |
| 53 | const defaultSleepDuration = 100 * time.Millisecond |
| 54 | sleepDuration := defaultSleepDuration |
| 55 | |
| 56 | slog.DebugContext(ctx, "watching for changes in the database", slog.Any("tenant_id", tenantID), slog.Any("snapshot", snap)) |
| 57 | // Decode snapshot token |
| 58 | // Decode the snapshot value. |
| 59 | // The snapshot value represents a point in the history of the database. |
| 60 | st, err := snapshot.EncodedToken{Value: snap}.Decode() |
| 61 | if err != nil { |
| 62 | // If there is an error in decoding the snapshot, send the error and return. |
| 63 | errs <- err |
| 64 | // Log decode error |
| 65 | slog.Error("failed to decode snapshot", slog.Any("error", err)) |
| 66 | // Return channels |
| 67 | return changes, errs |
| 68 | } |
| 69 | |
| 70 | // Start a goroutine to watch for changes in the database. |
| 71 | go func() { |
| 72 | // Ensure to close the channels when we're done. |
| 73 | defer close(changes) |
| 74 | defer close(errs) |
| 75 | |
| 76 | // Get the transaction ID from the snapshot. |
| 77 | cr := st.(snapshot.Token).Value.Uint |
| 78 | |
| 79 | // Continuously watch for changes. |
| 80 | for { |
| 81 | // Get the list of recent transaction IDs. |
| 82 | recentIDs, err := w.getRecentXIDs(ctx, cr, tenantID) |
| 83 | if err != nil { |
| 84 | // If there is an error in getting recent transaction IDs, send the error and return. |
| 85 | // Log transaction error |
| 86 | slog.Error("error getting recent transaction", slog.Any("error", err)) |
| 87 | errs <- err |
| 88 | return |
| 89 | } |
| 90 | |
| 91 | // Process each recent transaction ID. |
| 92 | for _, id := range recentIDs { |
| 93 | // Get the changes in the database associated with the current transaction ID. |
| 94 | updates, err := w.getChanges(ctx, id, tenantID) |
| 95 | if err != nil { |
| 96 | // If there is an error in getting the changes, send the error and return. |
| 97 | slog.ErrorContext(ctx, "failed to get changes for transaction", slog.Any("id", id), slog.Any("error", err)) |
| 98 | errs <- err |
| 99 | return |
| 100 | } |
| 101 | |
| 102 | // Send the changes, but respect the context cancellation. |
| 103 | select { |
nothing calls this directly
no test coverage detected