Skip to content

Commit b5e1b6e

Browse files
alexluongalexluong
authored
feat: Remember deleted entity resources (#121)
* feat: Delete destination * test: Fix issue from parallel test in e2e suite * feat: Delete tenant * fix: Error case for tenant delete ops * fix: Address delete & recreate case * fix: Tenant re-create after deleted
1 parent 058f1f7 commit b5e1b6e

File tree

10 files changed

+243
-53
lines changed

10 files changed

+243
-53
lines changed

‎cmd/e2e/api_test.go‎

Lines changed: 53 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@ import (
88
)
99

1010
func (suite *basicSuite) TestHealthzAPI() {
11-
suite.T().Parallel()
1211
tests := []APITest{
1312
{
1413
Name: "GET /healthz",
@@ -27,7 +26,6 @@ func (suite *basicSuite) TestHealthzAPI() {
2726
}
2827

2928
func (suite *basicSuite) TestTenantsAPI() {
30-
suite.T().Parallel()
3129
tenantID := uuid.New().String()
3230
sampleDestinationID := uuid.New().String()
3331
tests := []APITest{
@@ -216,12 +214,64 @@ func (suite *basicSuite) TestTenantsAPI() {
216214
},
217215
},
218216
},
217+
{
218+
Name: "DELETE /:tenantID",
219+
Request: suite.AuthRequest(httpclient.Request{
220+
Method: httpclient.MethodDELETE,
221+
Path: "/" + tenantID,
222+
}),
223+
Expected: APITestExpectation{
224+
Match: &httpclient.Response{
225+
StatusCode: http.StatusOK,
226+
},
227+
},
228+
},
229+
{
230+
Name: "GET /:tenantID",
231+
Request: suite.AuthRequest(httpclient.Request{
232+
Method: httpclient.MethodGET,
233+
Path: "/" + tenantID,
234+
}),
235+
Expected: APITestExpectation{
236+
Match: &httpclient.Response{
237+
StatusCode: http.StatusNotFound,
238+
},
239+
},
240+
},
241+
{
242+
Name: "POST /:tenantID/destinations",
243+
Request: suite.AuthRequest(httpclient.Request{
244+
Method: httpclient.MethodPOST,
245+
Path: "/" + tenantID + "/destinations",
246+
}),
247+
Expected: APITestExpectation{
248+
Match: &httpclient.Response{
249+
StatusCode: http.StatusNotFound,
250+
},
251+
},
252+
},
253+
{
254+
Name: "PUT /:tenantID should override deleted tenant",
255+
Request: suite.AuthRequest(httpclient.Request{
256+
Method: httpclient.MethodPUT,
257+
Path: "/" + tenantID,
258+
}),
259+
Expected: APITestExpectation{
260+
Match: &httpclient.Response{
261+
StatusCode: http.StatusCreated,
262+
Body: map[string]interface{}{
263+
"id": tenantID,
264+
"destinations_count": 0,
265+
"topics": []string{},
266+
},
267+
},
268+
},
269+
},
219270
}
220271
suite.RunAPITests(suite.T(), tests)
221272
}
222273

223274
func (suite *basicSuite) TestDestinationsAPI() {
224-
suite.T().Parallel()
225275
tenantID := uuid.New().String()
226276
sampleDestinationID := uuid.New().String()
227277
tests := []APITest{
@@ -572,7 +622,6 @@ func (suite *basicSuite) TestDestinationsAPI() {
572622
}
573623

574624
func (suite *basicSuite) TestDestinationsListAPI() {
575-
suite.T().Parallel()
576625
tenantID := uuid.New().String()
577626
tests := []APITest{
578627
{
@@ -739,7 +788,6 @@ func (suite *basicSuite) TestDestinationsListAPI() {
739788
}
740789

741790
func (suite *basicSuite) TestDestinationEnableDisableAPI() {
742-
suite.T().Parallel()
743791
tenantID := uuid.New().String()
744792
sampleDestinationID := uuid.New().String()
745793
tests := []APITest{
@@ -880,7 +928,6 @@ func (suite *basicSuite) TestDestinationEnableDisableAPI() {
880928
}
881929

882930
func (suite *basicSuite) TestTopicsAPI() {
883-
suite.T().Parallel()
884931
tests := []APITest{
885932
{
886933
Name: "GET /topics",

‎cmd/e2e/publish_test.go‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,6 @@ import (
1010
)
1111

1212
func (suite *basicSuite) TestPublishAPI() {
13-
suite.T().Parallel()
1413
tenantID := uuid.New().String()
1514
sampleDestinationID := uuid.New().String()
1615
eventIDs := []string{uuid.New().String(), uuid.New().String()}

‎cmd/e2e/suites_test.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,7 @@ func (s *e2eSuite) AuthRequest(req httpclient.Request) httpclient.Request {
5151
}
5252

5353
func (suite *e2eSuite) RunAPITests(t *testing.T, tests []APITest) {
54+
t.Helper()
5455
for _, test := range tests {
5556
t.Run(test.Name, func(t *testing.T) {
5657
test.Run(t, suite.client)
@@ -71,6 +72,8 @@ type APITestExpectation struct {
7172
}
7273

7374
func (test *APITest) Run(t *testing.T, client httpclient.Client) {
75+
t.Helper()
76+
7477
if test.Delay > 0 {
7578
time.Sleep(test.Delay)
7679
}
@@ -128,6 +131,7 @@ func (s *basicSuite) TearDownSuite() {
128131
}
129132

130133
func TestBasicSuite(t *testing.T) {
134+
t.Parallel()
131135
if testing.Short() {
132136
t.Skip("skipping e2e test")
133137
}

‎internal/models/destination.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,10 @@ func (d *Destination) parseRedisHash(cmd *redis.MapStringStringCmd, cipher Ciphe
4343
if len(hash) == 0 {
4444
return redis.Nil
4545
}
46+
// Check for deleted resource before scanning
47+
if _, exists := hash["deleted_at"]; exists {
48+
return ErrDestinationDeleted
49+
}
4650
if err = cmd.Scan(d); err != nil {
4751
return err
4852
}

‎internal/models/entity.go‎

Lines changed: 70 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"fmt"
77
"slices"
88
"sort"
9+
"time"
910

1011
"github.com/hookdeck/outpost/internal/redis"
1112
)
@@ -26,7 +27,11 @@ type EntityStore interface {
2627
}
2728

2829
var (
30+
ErrTenantNotFound = errors.New("tenant does not exist")
31+
ErrTenantDeleted = errors.New("tenant has been deleted")
2932
ErrDuplicateDestination = errors.New("destination already exists")
33+
ErrDestinationNotFound = errors.New("destination does not exist")
34+
ErrDestinationDeleted = errors.New("destination has been deleted")
3035
)
3136

3237
func redisTenantID(tenantID string) string {
@@ -89,26 +94,50 @@ func (s *entityStoreImpl) RetrieveTenant(ctx context.Context, tenantID string) (
8994
}
9095

9196
func (s *entityStoreImpl) UpsertTenant(ctx context.Context, tenant Tenant) error {
92-
return s.redisClient.HSet(ctx, redisTenantID(tenant.ID), tenant).Err()
97+
key := redisTenantID(tenant.ID)
98+
99+
_, err := s.redisClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
100+
// Support overriding deleted resources
101+
pipe.Persist(ctx, key)
102+
pipe.HDel(ctx, key, "deleted_at")
103+
104+
// Set tenant data
105+
pipe.HSet(ctx, key, tenant)
106+
return nil
107+
})
108+
109+
return err
93110
}
94111

95112
func (s *entityStoreImpl) DeleteTenant(ctx context.Context, tenantID string) error {
96113
maxRetries := 100
97114

115+
if exists, err := s.redisClient.Exists(ctx, redisTenantID(tenantID)).Result(); err != nil {
116+
return err
117+
} else if exists == 0 {
118+
return ErrTenantNotFound
119+
}
120+
98121
txf := func(tx *redis.Tx) error {
99122
destinationIDs, err := s.redisClient.HKeys(ctx, redisTenantDestinationSummaryKey(tenantID)).Result()
100123
if err != nil {
101124
return err
102125
}
103-
_, err = s.redisClient.TxPipelined(ctx, func(r redis.Pipeliner) error {
126+
if _, err := s.redisClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
127+
now := time.Now()
104128
for _, destinationID := range destinationIDs {
105-
r.Del(ctx, redisDestinationID(destinationID, tenantID))
129+
s.deleteDestinationOperation(ctx, pipe, redisDestinationID(destinationID, tenantID), now)
106130
}
107-
r.Del(ctx, redisTenantDestinationSummaryKey(tenantID))
108-
r.Del(ctx, redisTenantID(tenantID))
131+
pipe.Del(ctx, redisTenantDestinationSummaryKey(tenantID))
132+
tenantKey := redisTenantID(tenantID)
133+
pipe.Del(ctx, tenantKey)
134+
pipe.HSet(ctx, tenantKey, "deleted_at", now)
135+
pipe.Expire(ctx, tenantKey, 7*24*time.Hour)
109136
return nil
110-
})
111-
return err
137+
}); err != nil {
138+
return err
139+
}
140+
return nil
112141
}
113142

114143
for i := 0; i < maxRetries; i++ {
@@ -208,12 +237,13 @@ func (s *entityStoreImpl) RetrieveDestination(ctx context.Context, tenantID, des
208237

209238
func (m *entityStoreImpl) CreateDestination(ctx context.Context, destination Destination) error {
210239
key := redisDestinationID(destination.ID, destination.TenantID)
211-
destinationExists, err := m.redisClient.Exists(ctx, key).Result()
212-
if err != nil {
240+
// Check if destination exists
241+
if fields, err := m.redisClient.HGetAll(ctx, key).Result(); err != nil {
213242
return err
214-
}
215-
if destinationExists > 0 {
216-
return ErrDuplicateDestination
243+
} else if len(fields) > 0 {
244+
if _, isDeleted := fields["deleted_at"]; !isDeleted {
245+
return ErrDuplicateDestination
246+
}
217247
}
218248
return m.UpsertDestination(ctx, destination)
219249
}
@@ -229,6 +259,10 @@ func (m *entityStoreImpl) UpsertDestination(ctx context.Context, destination Des
229259
if err != nil {
230260
return err
231261
}
262+
// Support overriding deleted resources
263+
r.Persist(ctx, key)
264+
r.HDel(ctx, key, "deleted_at")
265+
// Set the new destination values
232266
r.HSet(ctx, key, "id", destination.ID)
233267
r.HSet(ctx, key, "type", destination.Type)
234268
r.HSet(ctx, key, "topics", &destination.Topics)
@@ -247,13 +281,30 @@ func (m *entityStoreImpl) UpsertDestination(ctx context.Context, destination Des
247281
}
248282

249283
func (s *entityStoreImpl) DeleteDestination(ctx context.Context, tenantID, destinationID string) error {
250-
_, err := s.redisClient.TxPipelined(ctx, func(r redis.Pipeliner) error {
251-
if err := r.HDel(ctx, redisTenantDestinationSummaryKey(tenantID), destinationID).Err(); err != nil {
252-
return err
253-
}
254-
return r.Del(ctx, redisDestinationID(destinationID, tenantID)).Err()
255-
})
256-
return err
284+
key := redisDestinationID(destinationID, tenantID)
285+
summaryKey := redisTenantDestinationSummaryKey(tenantID)
286+
287+
// Check if destination exists
288+
if exists, err := s.redisClient.Exists(ctx, key).Result(); err != nil {
289+
return err
290+
} else if exists == 0 {
291+
return ErrDestinationNotFound
292+
}
293+
294+
pipe := s.redisClient.Pipeline()
295+
pipe.HDel(ctx, summaryKey, destinationID)
296+
s.deleteDestinationOperation(ctx, pipe, key, time.Now())
297+
if _, err := pipe.Exec(ctx); err != nil {
298+
return err
299+
}
300+
301+
return nil
302+
}
303+
304+
func (s *entityStoreImpl) deleteDestinationOperation(ctx context.Context, pipe redis.Pipeliner, key string, ts time.Time) {
305+
pipe.Del(ctx, key)
306+
pipe.HSet(ctx, key, "deleted_at", ts)
307+
pipe.Expire(ctx, key, 7*24*time.Hour)
257308
}
258309

259310
func (s *entityStoreImpl) MatchEvent(ctx context.Context, event Event) ([]DestinationSummary, error) {

0 commit comments

Comments
 (0)