ai cooking
paralelize and remove actowiz (#398)
* paralelize and remove actowiz
* error group fast cancel
---------
Co-authored-by: paul miller <paul.miller>
authored by
Paul Miller
paul miller
and committed by
GitHub
9b937d40
331216b4
+184
-31
+3
-1
internal/albertsons/cache.go
+3
-1
internal/albertsons/cache.go
···
8
8
"log/slog"
9
9
10
10
"careme/internal/cache"
11
+
"careme/internal/sitemapfetch"
12
+
11
13
locationtypes "careme/internal/locations/types"
12
-
"careme/internal/sitemapfetch"
13
14
14
15
"github.com/samber/lo"
15
16
lop "github.com/samber/lo/parallel"
···
52
53
}
53
54
54
55
// expensive. Just save a smaller map of centroids
56
+
// see branch cachezipcodes. Maybe time for a real db instead of blob
55
57
summaries := lop.Map(keys, func(key string, _ int) *StoreSummary {
56
58
reader, err := c.Get(ctx, StoreCachePrefix+key)
57
59
if err != nil {
+50
-18
internal/locations/storage.go
+50
-18
internal/locations/storage.go
···
25
25
locationtypes "careme/internal/locations/types"
26
26
27
27
"github.com/samber/lo"
28
+
"golang.org/x/sync/errgroup"
28
29
)
29
30
30
31
type locationStorage struct {
···
58
59
ZipCentroidByZIP(zip string) (locationtypes.ZipCentroid, bool)
59
60
}
60
61
62
+
type locationBackendFactory func(context.Context) (locationBackend, error)
63
+
61
64
// bad for rural areas if zip code is huge?
62
65
const (
63
66
maxLocationDistanceMiles = 20.0
···
75
78
}
76
79
77
80
ctx := context.Background()
78
-
type locationBackendFactory func() (locationBackend, error)
79
81
backendfactories := []locationBackendFactory{
80
-
func() (locationBackend, error) { return kroger.FromConfig(cfg) },
81
-
func() (locationBackend, error) { return walmart.NewClient(cfg.Walmart) },
82
-
func() (locationBackend, error) { return aldi.NewLocationBackendFromConfig(ctx, cfg, centroids) },
83
-
func() (locationBackend, error) { return wholefoods.NewLocationBackendFromConfig(ctx, cfg, centroids) },
84
-
func() (locationBackend, error) { return albertsons.NewLocationBackendFromConfig(ctx, cfg, centroids) },
85
-
func() (locationBackend, error) { return publix.NewLocationBackendFromConfig(ctx, cfg, centroids) },
86
-
func() (locationBackend, error) { return heb.NewLocationBackendFromConfig(ctx, cfg, centroids) },
82
+
func(context.Context) (locationBackend, error) { return kroger.FromConfig(cfg) },
83
+
func(context.Context) (locationBackend, error) { return walmart.NewClient(cfg.Walmart) },
84
+
func(ctx context.Context) (locationBackend, error) {
85
+
return aldi.NewLocationBackendFromConfig(ctx, cfg, centroids)
86
+
},
87
+
func(ctx context.Context) (locationBackend, error) {
88
+
return wholefoods.NewLocationBackendFromConfig(ctx, cfg, centroids)
89
+
},
90
+
func(ctx context.Context) (locationBackend, error) {
91
+
return albertsons.NewLocationBackendFromConfig(ctx, cfg, centroids)
92
+
},
93
+
func(ctx context.Context) (locationBackend, error) {
94
+
return publix.NewLocationBackendFromConfig(ctx, cfg, centroids)
95
+
},
96
+
func(ctx context.Context) (locationBackend, error) {
97
+
return heb.NewLocationBackendFromConfig(ctx, cfg, centroids)
98
+
},
87
99
}
88
100
89
-
backends := make([]locationBackend, 0, len(backendfactories))
90
-
for i, factory := range backendfactories {
91
-
backend, err := factory()
92
-
if err != nil {
93
-
if locationtypes.IsDisabledBackendError(err) {
94
-
continue
95
-
}
96
-
return nil, fmt.Errorf("failed to initialize location backend %d: %w", i, err)
97
-
}
98
-
backends = append(backends, backend)
101
+
backends, err := initializeLocationBackends(ctx, backendfactories)
102
+
if err != nil {
103
+
return nil, err
99
104
}
100
105
101
106
return &locationStorage{
···
103
108
zipCentroids: centroids,
104
109
cache: c,
105
110
}, nil
111
+
}
112
+
113
+
func initializeLocationBackends(ctx context.Context, factories []locationBackendFactory) ([]locationBackend, error) {
114
+
g, ctx := errgroup.WithContext(ctx)
115
+
results := make(chan locationBackend, len(factories))
116
+
for i, factory := range factories {
117
+
i, factory := i, factory
118
+
g.Go(func() error {
119
+
start := time.Now()
120
+
backend, err := factory(ctx)
121
+
if err != nil {
122
+
if locationtypes.IsDisabledBackendError(err) {
123
+
return nil
124
+
}
125
+
return fmt.Errorf("failed to initialize location backend %d: %w", i, err)
126
+
}
127
+
slog.InfoContext(ctx, "initialized location backend", "backend", fmt.Sprintf("%T", backend), "latencyMS", time.Since(start).Milliseconds())
128
+
results <- backend
129
+
return nil
130
+
})
131
+
}
132
+
if err := g.Wait(); err != nil {
133
+
return nil, err
134
+
}
135
+
close(results)
136
+
137
+
return lo.ChannelToSlice(results), nil
106
138
}
107
139
108
140
func (l *locationStorage) HasInventory(locationID string) bool {
+129
internal/locations/storage_test.go
+129
internal/locations/storage_test.go
···
7
7
"fmt"
8
8
"strings"
9
9
"testing"
10
+
"time"
10
11
11
12
cachepkg "careme/internal/cache"
12
13
)
14
+
15
+
type namedBackend struct {
16
+
id string
17
+
}
18
+
19
+
func (b namedBackend) GetLocationByID(context.Context, string) (*Location, error) {
20
+
return nil, fmt.Errorf("not implemented")
21
+
}
22
+
23
+
func (b namedBackend) GetLocationsByZip(context.Context, string) ([]Location, error) {
24
+
return nil, nil
25
+
}
26
+
27
+
func (b namedBackend) IsID(string) bool {
28
+
return false
29
+
}
30
+
31
+
func (b namedBackend) HasInventory(string) bool {
32
+
return false
33
+
}
34
+
35
+
func TestInitializeLocationBackendsRunsFactoriesInParallelAndCollectsBackends(t *testing.T) {
36
+
started := make(chan string, 2)
37
+
release := make(chan struct{})
38
+
39
+
factories := []locationBackendFactory{
40
+
func(context.Context) (locationBackend, error) {
41
+
started <- "first"
42
+
<-release
43
+
return namedBackend{id: "first"}, nil
44
+
},
45
+
func(context.Context) (locationBackend, error) {
46
+
started <- "second"
47
+
<-release
48
+
return namedBackend{id: "second"}, nil
49
+
},
50
+
}
51
+
52
+
type result struct {
53
+
backends []locationBackend
54
+
err error
55
+
}
56
+
done := make(chan result, 1)
57
+
go func() {
58
+
backends, err := initializeLocationBackends(context.Background(), factories)
59
+
done <- result{backends: backends, err: err}
60
+
}()
61
+
62
+
for i := 0; i < len(factories); i++ {
63
+
select {
64
+
case <-started:
65
+
case <-time.After(200 * time.Millisecond):
66
+
t.Fatal("expected all backend factories to start before any finished")
67
+
}
68
+
}
69
+
70
+
close(release)
71
+
72
+
select {
73
+
case result := <-done:
74
+
if result.err != nil {
75
+
t.Fatalf("initializeLocationBackends returned error: %v", result.err)
76
+
}
77
+
if len(result.backends) != 2 {
78
+
t.Fatalf("expected 2 backends, got %d", len(result.backends))
79
+
}
80
+
81
+
gotIDs := make(map[string]bool, len(result.backends))
82
+
for _, backend := range result.backends {
83
+
named, ok := backend.(namedBackend)
84
+
if !ok {
85
+
t.Fatalf("expected backend type namedBackend, got %T", backend)
86
+
}
87
+
gotIDs[named.id] = true
88
+
}
89
+
if !gotIDs["first"] || !gotIDs["second"] {
90
+
t.Fatalf("expected both backends to be returned, got %v", gotIDs)
91
+
}
92
+
case <-time.After(time.Second):
93
+
t.Fatal("initializeLocationBackends did not finish")
94
+
}
95
+
}
96
+
97
+
func TestInitializeLocationBackendsCancelsOtherFactoriesOnError(t *testing.T) {
98
+
started := make(chan struct{})
99
+
releaseErr := make(chan struct{})
100
+
canceled := make(chan struct{}, 1)
101
+
102
+
factories := []locationBackendFactory{
103
+
func(context.Context) (locationBackend, error) {
104
+
close(started)
105
+
<-releaseErr
106
+
return nil, errors.New("boom")
107
+
},
108
+
func(ctx context.Context) (locationBackend, error) {
109
+
<-started
110
+
<-ctx.Done()
111
+
canceled <- struct{}{}
112
+
return nil, ctx.Err()
113
+
},
114
+
}
115
+
116
+
done := make(chan error, 1)
117
+
go func() {
118
+
_, err := initializeLocationBackends(context.Background(), factories)
119
+
done <- err
120
+
}()
121
+
122
+
close(releaseErr)
123
+
124
+
select {
125
+
case err := <-done:
126
+
if err == nil {
127
+
t.Fatal("initializeLocationBackends error = nil, want error")
128
+
}
129
+
if got, want := err.Error(), "failed to initialize location backend 0: boom"; got != want {
130
+
t.Fatalf("initializeLocationBackends error = %q, want %q", got, want)
131
+
}
132
+
case <-time.After(time.Second):
133
+
t.Fatal("initializeLocationBackends did not return after error")
134
+
}
135
+
136
+
select {
137
+
case <-canceled:
138
+
case <-time.After(time.Second):
139
+
t.Fatal("expected sibling factory context to be canceled")
140
+
}
141
+
}
13
142
14
143
func TestGetLocationByIDUsesCache(t *testing.T) {
15
144
client := newFakeLocationClient()
+2
-3
internal/recipes/staples.go
+2
-3
internal/recipes/staples.go
···
5
5
"fmt"
6
6
"testing"
7
7
8
-
"careme/internal/actowiz"
9
8
"careme/internal/config"
10
9
"careme/internal/kroger"
11
10
"careme/internal/walmart"
···
85
84
func defaultStaplesBackends(krogerClient kroger.ClientWithResponsesInterface) []backendStaplesProvider {
86
85
return []backendStaplesProvider{
87
86
kroger.NewStaplesProvider(krogerClient),
88
-
actowiz.NewStaplesProvider(),
87
+
// actowiz.NewStaplesProvider(),
89
88
wholefoods.NewStaplesProvider(wholefoods.NewClient(nil)),
90
89
walmart.NewStaplesProvider(),
91
90
}
···
94
93
func defaultIdentityProviders() []identityProvider {
95
94
return []identityProvider{
96
95
kroger.NewIdentityProvider(),
97
-
actowiz.NewIdentityProvider(),
96
+
// actowiz.NewIdentityProvider(),
98
97
wholefoods.NewIdentityProvider(),
99
98
walmart.NewIdentityProvider(),
100
99
}
-9
internal/recipes/staples_test.go
-9
internal/recipes/staples_test.go
···
6
6
"testing"
7
7
"time"
8
8
9
-
"careme/internal/actowiz"
10
9
"careme/internal/cache"
11
10
"careme/internal/kroger"
12
11
"careme/internal/locations"
···
122
121
}
123
122
if krogerProvider.calls != 0 || wholeFoodsProvider.calls != 1 {
124
123
t.Fatalf("expected whole foods provider to be selected, got kroger=%d wholefoods=%d", krogerProvider.calls, wholeFoodsProvider.calls)
125
-
}
126
-
}
127
-
128
-
func TestStaplesSignatureForLocation_UsesSafewayProvider(t *testing.T) {
129
-
got := staplesSignatureForLocation("safeway_1234")
130
-
want := actowiz.NewIdentityProvider().Signature()
131
-
if got != want {
132
-
t.Fatalf("unexpected signature: got %q want %q", got, want)
133
124
}
134
125
}
135
126