This repository has been archived by the owner on Apr 2, 2024. It is now read-only.
generated from mrz1836/go-template
-
-
Notifications
You must be signed in to change notification settings - Fork 6
/
cron_job_definitions.go
98 lines (83 loc) · 2.68 KB
/
cron_job_definitions.go
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
package bux
import (
"context"
"errors"
"time"
"github.com/mrz1836/go-datastore"
)
// taskCleanupDraftTransactions will clean up all old expired draft transactions
func taskCleanupDraftTransactions(ctx context.Context, client *Client) error {
client.Logger().Info().Msg("running cleanup draft transactions task...")
// Construct an empty model
var models []DraftTransaction
conditions := map[string]interface{}{
statusField: DraftStatusDraft,
// todo: add DB condition for date "expires_at": map[string]interface{}{"$lte": time.Now()},
}
queryParams := &datastore.QueryParams{
Page: 1,
PageSize: 20,
OrderByField: idField,
SortDirection: datastore.SortAsc,
}
// Get the records
if err := getModels(
ctx, client.Datastore(),
&models, conditions, queryParams, defaultDatabaseReadTimeout,
); err != nil {
if errors.Is(err, datastore.ErrNoResults) {
return nil
}
return err
}
// Loop and update
var err error
timeNow := time.Now().UTC()
for index := range models {
if timeNow.After(models[index].ExpiresAt) {
models[index].enrich(ModelDraftTransaction, WithClient(client))
models[index].Status = DraftStatusExpired
if err = models[index].Save(ctx); err != nil {
return err
}
}
}
return nil
}
// taskProcessIncomingTransactions will process any incoming transactions found
func taskProcessIncomingTransactions(ctx context.Context, client *Client) error {
client.Logger().Info().Msg("running process incoming transaction(s) task...")
err := processIncomingTransactions(ctx, client.Logger(), 10, WithClient(client))
if err == nil || errors.Is(err, datastore.ErrNoResults) {
return nil
}
return err
}
// taskBroadcastTransactions will broadcast any transactions
func taskBroadcastTransactions(ctx context.Context, client *Client) error {
client.Logger().Info().Msg("running broadcast transaction(s) task...")
err := processBroadcastTransactions(ctx, 1000, WithClient(client))
if err == nil || errors.Is(err, datastore.ErrNoResults) {
return nil
}
return err
}
// taskSyncTransactions will sync any transactions
func taskSyncTransactions(ctx context.Context, client *Client) error {
logClient := client.Logger()
logClient.Info().Msg("running sync transaction(s) task...")
// Prevent concurrent running
unlock, err := newWriteLock(
ctx, lockKeyProcessSyncTx, client.Cachestore(),
)
defer unlock()
if err != nil {
logClient.Warn().Msg("cannot run sync transaction(s) task, previous run is not complete yet...")
return nil //nolint:nilerr // previous run is not complete yet
}
err = processSyncTransactions(ctx, 100, WithClient(client))
if err == nil || errors.Is(err, datastore.ErrNoResults) {
return nil
}
return err
}