Skip to content

Commit f5ad75b

Browse files
alnralnr
committed
feat: pause/resume event streams
1 parent a0c7075 commit f5ad75b

10 files changed

Lines changed: 249 additions & 6 deletions

File tree

‎cmd/cloudx/eventstreams/flags.go‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,9 +12,28 @@ import (
1212
"github.com/ory/client-go"
1313
)
1414

15+
// Event stream statuses. A paused stream does not forward any events until it
16+
// is set back to active.
17+
const (
18+
StatusActive = "active"
19+
StatusPaused = "paused"
20+
)
21+
1522
type streamConfig client.CreateEventStreamBody
1623

1724
func (c *streamConfig) Validate() error {
25+
// The status flag is optional. An empty value is normalized to nil so the
26+
// server keeps the current status (on update) or applies its default (on create).
27+
if c.Status != nil {
28+
switch *c.Status {
29+
case "":
30+
c.Status = nil
31+
case StatusActive, StatusPaused:
32+
default:
33+
return fmt.Errorf(`flag --status must be one of %q or %q`, StatusActive, StatusPaused)
34+
}
35+
}
36+
1837
switch c.Type {
1938
case "":
2039
return fmt.Errorf("flag --type must be set")
@@ -51,9 +70,27 @@ func (c *streamConfig) Validate() error {
5170
return nil
5271
}
5372

73+
// toSetBody maps the shared stream config onto the update (set) request body.
74+
// The two bodies are no longer convertible by type assertion: SetEventStreamBody.Type
75+
// is a pointer (optional on update) whereas CreateEventStreamBody.Type is required.
76+
func (c streamConfig) toSetBody() client.SetEventStreamBody {
77+
body := client.SetEventStreamBody{
78+
HttpsEndpoint: c.HttpsEndpoint,
79+
RoleArn: c.RoleArn,
80+
Status: c.Status,
81+
TopicArn: c.TopicArn,
82+
}
83+
if c.Type != "" {
84+
t := c.Type
85+
body.Type = &t
86+
}
87+
return body
88+
}
89+
5490
func registerStreamConfigFlags(f *pflag.FlagSet, c *streamConfig) {
5591
f.StringVar(&c.Type, "type", "", `The type of the event stream destination. Supported values are "sns" for AWS SNS topics and "https" for generic HTTPS endpoints.`)
5692
c.RoleArn = f.String("aws-iam-role-arn", "", "The ARN of the AWS IAM role to assume when publishing messages to the SNS topic.")
5793
c.TopicArn = f.String("aws-sns-topic-arn", "", "The ARN of the AWS SNS topic.")
5894
c.HttpsEndpoint = f.String("https-endpoint", "", "The URL of the HTTPS endpoint.")
95+
c.Status = f.String("status", "", fmt.Sprintf("The status of the event stream. Supported values are %q and %q. Defaults to %q.", StatusActive, StatusPaused, StatusActive))
5996
}

‎cmd/cloudx/eventstreams/flags_test.go‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,84 @@
1+
// Copyright © 2024 Ory Corp
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package eventstreams
5+
6+
import (
7+
"testing"
8+
9+
"github.com/stretchr/testify/assert"
10+
"github.com/stretchr/testify/require"
11+
)
12+
13+
func ptr(s string) *string { return &s }
14+
15+
func TestStreamConfigValidate(t *testing.T) {
16+
t.Parallel()
17+
18+
base := func() streamConfig {
19+
return streamConfig{
20+
Type: "https",
21+
HttpsEndpoint: ptr("https://example.com/webhook"),
22+
}
23+
}
24+
25+
t.Run("accepts a valid active status", func(t *testing.T) {
26+
c := base()
27+
c.Status = ptr(StatusActive)
28+
require.NoError(t, c.Validate())
29+
assert.Equal(t, StatusActive, *c.Status)
30+
})
31+
32+
t.Run("accepts a valid paused status", func(t *testing.T) {
33+
c := base()
34+
c.Status = ptr(StatusPaused)
35+
require.NoError(t, c.Validate())
36+
assert.Equal(t, StatusPaused, *c.Status)
37+
})
38+
39+
t.Run("normalizes an empty status to nil so the server default applies", func(t *testing.T) {
40+
c := base()
41+
c.Status = ptr("")
42+
require.NoError(t, c.Validate())
43+
assert.Nil(t, c.Status)
44+
})
45+
46+
t.Run("rejects an unknown status", func(t *testing.T) {
47+
c := base()
48+
c.Status = ptr("frozen")
49+
assert.ErrorContains(t, c.Validate(), "--status")
50+
})
51+
52+
t.Run("status is optional when unset", func(t *testing.T) {
53+
c := base()
54+
require.NoError(t, c.Validate())
55+
assert.Nil(t, c.Status)
56+
})
57+
}
58+
59+
func TestStreamConfigToSetBody(t *testing.T) {
60+
t.Parallel()
61+
62+
t.Run("maps all fields including the required type as a pointer", func(t *testing.T) {
63+
c := streamConfig{
64+
Type: "https",
65+
HttpsEndpoint: ptr("https://example.com/webhook"),
66+
Status: ptr(StatusPaused),
67+
}
68+
body := c.toSetBody()
69+
require.NotNil(t, body.Type)
70+
assert.Equal(t, "https", *body.Type)
71+
require.NotNil(t, body.HttpsEndpoint)
72+
assert.Equal(t, "https://example.com/webhook", *body.HttpsEndpoint)
73+
require.NotNil(t, body.Status)
74+
assert.Equal(t, StatusPaused, *body.Status)
75+
})
76+
77+
t.Run("leaves type nil when unset so the current type is kept", func(t *testing.T) {
78+
c := streamConfig{Status: ptr(StatusActive)}
79+
body := c.toSetBody()
80+
assert.Nil(t, body.Type)
81+
require.NotNil(t, body.Status)
82+
assert.Equal(t, StatusActive, *body.Status)
83+
})
84+
}

‎cmd/cloudx/eventstreams/output.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,14 @@ type (
1515
)
1616

1717
func (output) Header() []string {
18-
return []string{"ID", "TYPE", "IAM_ROLE_ARN", "SNS_TOPIC_ARN", "HTTPS_ENDPOINT"}
18+
return []string{"ID", "TYPE", "STATUS", "IAM_ROLE_ARN", "SNS_TOPIC_ARN", "HTTPS_ENDPOINT"}
1919
}
2020

2121
func (o output) Columns() []string {
2222
return []string{
2323
coalesce(o.Id),
2424
coalesce(o.Type),
25+
coalesce(o.Status),
2526
coalesce(o.RoleArn),
2627
coalesce(o.TopicArn),
2728
coalesce(o.HttpsEndpoint.Get()),

‎cmd/cloudx/eventstreams/status.go‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
// Copyright © 2024 Ory Corp
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package eventstreams
5+
6+
import (
7+
"fmt"
8+
9+
"github.com/spf13/cobra"
10+
11+
"github.com/ory/cli/cmd/cloudx/client"
12+
cloud "github.com/ory/client-go"
13+
"github.com/ory/x/cmdx"
14+
)
15+
16+
func NewPauseEventStreamCmd() *cobra.Command {
17+
return newSetStatusCmd("pause", StatusPaused, "Pause the event stream with the given ID", "A paused event stream does not forward any events until it is resumed.")
18+
}
19+
20+
func NewResumeEventStreamCmd() *cobra.Command {
21+
return newSetStatusCmd("resume", StatusActive, "Resume the event stream with the given ID", "Resuming a paused event stream makes it forward events again.")
22+
}
23+
24+
func newSetStatusCmd(verb, status, short, long string) *cobra.Command {
25+
cmd := &cobra.Command{
26+
Use: "event-stream <id> [--project=PROJECT_ID]",
27+
Args: cobra.ExactArgs(1),
28+
Short: short,
29+
Long: short + "\n\n" + long,
30+
RunE: func(cmd *cobra.Command, args []string) error {
31+
ctx := cmd.Context()
32+
33+
h, err := client.NewCobraCommandHelper(cmd)
34+
if err != nil {
35+
return err
36+
}
37+
38+
projectID, err := h.ProjectID()
39+
if err != nil {
40+
return cmdx.PrintOpenAPIError(cmd, err)
41+
}
42+
streamID := args[0]
43+
44+
stream, err := h.UpdateEventStream(ctx, projectID, streamID, cloud.SetEventStreamBody{Status: &status})
45+
if err != nil {
46+
return cmdx.PrintOpenAPIError(cmd, err)
47+
}
48+
49+
_, _ = fmt.Fprintf(h.VerboseErrWriter, "Event stream %sd successfully!\n", verb)
50+
cmdx.PrintRow(cmd, output(*stream))
51+
return nil
52+
},
53+
}
54+
55+
client.RegisterProjectFlag(cmd.Flags())
56+
client.RegisterWorkspaceFlag(cmd.Flags())
57+
cmdx.RegisterFormatFlags(cmd.Flags())
58+
return cmd
59+
}

‎cmd/cloudx/eventstreams/update.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@ import (
99
"github.com/spf13/cobra"
1010

1111
"github.com/ory/cli/cmd/cloudx/client"
12-
cloud "github.com/ory/client-go"
1312

1413
"github.com/ory/x/cmdx"
1514
)
@@ -37,7 +36,7 @@ func NewUpdateEventStreamCmd() *cobra.Command {
3736
if err := c.Validate(); err != nil {
3837
return err
3938
}
40-
stream, err := h.UpdateEventStream(ctx, projectID, streamID, cloud.SetEventStreamBody(c))
39+
stream, err := h.UpdateEventStream(ctx, projectID, streamID, c.toSetBody())
4140
if err != nil {
4241
return cmdx.PrintOpenAPIError(cmd, err)
4342
}
@@ -49,7 +48,10 @@ func NewUpdateEventStreamCmd() *cobra.Command {
4948
}
5049

5150
client.RegisterProjectFlag(cmd.Flags())
51+
client.RegisterWorkspaceFlag(cmd.Flags())
5252
cmdx.RegisterFormatFlags(cmd.Flags())
5353

54+
registerStreamConfigFlags(cmd.Flags(), &c)
55+
5456
return cmd
5557
}

‎cmd/cloudx/pause.go‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
// Copyright © 2024 Ory Corp
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package cloudx
5+
6+
import (
7+
"github.com/spf13/cobra"
8+
9+
"github.com/ory/cli/cmd/cloudx/client"
10+
"github.com/ory/cli/cmd/cloudx/eventstreams"
11+
"github.com/ory/x/cmdx"
12+
)
13+
14+
func NewPauseCmd() *cobra.Command {
15+
cmd := &cobra.Command{
16+
Use: "pause",
17+
Short: "Pause Ory Network resources",
18+
}
19+
20+
cmd.AddCommand(
21+
eventstreams.NewPauseEventStreamCmd(),
22+
)
23+
24+
client.RegisterConfigFlag(cmd.PersistentFlags())
25+
client.RegisterYesFlag(cmd.PersistentFlags())
26+
cmdx.RegisterNoiseFlags(cmd.PersistentFlags())
27+
cmdx.RegisterJSONFormatFlags(cmd.PersistentFlags())
28+
return cmd
29+
}

‎cmd/cloudx/resume.go‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
// Copyright © 2024 Ory Corp
2+
// SPDX-License-Identifier: Apache-2.0
3+
4+
package cloudx
5+
6+
import (
7+
"github.com/spf13/cobra"
8+
9+
"github.com/ory/cli/cmd/cloudx/client"
10+
"github.com/ory/cli/cmd/cloudx/eventstreams"
11+
"github.com/ory/x/cmdx"
12+
)
13+
14+
func NewResumeCmd() *cobra.Command {
15+
cmd := &cobra.Command{
16+
Use: "resume",
17+
Short: "Resume Ory Network resources",
18+
}
19+
20+
cmd.AddCommand(
21+
eventstreams.NewResumeEventStreamCmd(),
22+
)
23+
24+
client.RegisterConfigFlag(cmd.PersistentFlags())
25+
client.RegisterYesFlag(cmd.PersistentFlags())
26+
cmdx.RegisterNoiseFlags(cmd.PersistentFlags())
27+
cmdx.RegisterJSONFormatFlags(cmd.PersistentFlags())
28+
return cmd
29+
}

‎cmd/root.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,9 +38,11 @@ func NewRootCmd() *cobra.Command {
3838
cloudx.NewOpenCmd(),
3939
cloudx.NewPatchCmd(),
4040
cloudx.NewParseCmd(),
41+
cloudx.NewPauseCmd(),
4142
cloudx.NewPerformCmd(),
4243
proxy.NewProxyCommand(),
4344
proxy.NewTunnelCommand(),
45+
cloudx.NewResumeCmd(),
4446
cloudx.NewUpdateCmd(),
4547
cloudx.NewValidateCmd(),
4648
cloudx.NewRevokeCmd(),

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ require (
1515
github.com/gofrs/uuid v4.4.0+incompatible
1616
github.com/gomarkdown/markdown v0.0.0-20260417124207-7d523f7318df
1717
github.com/hashicorp/go-retryablehttp v0.7.8
18-
github.com/ory/client-go v1.22.41
18+
github.com/ory/client-go v1.22.51
1919
github.com/ory/gochimp3 v0.0.0-20200417124117-ccd242db3655
2020
github.com/ory/graceful v0.2.0
2121
github.com/ory/herodot v0.10.9-0.20260330111132-da75ef0fbc22

‎go.sum‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -554,8 +554,8 @@ github.com/openzipkin/zipkin-go v0.4.3/go.mod h1:M9wCJZFWCo2RiY+o1eBCEMe0Dp2S5LD
554554
github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0=
555555
github.com/ory/analytics-go/v5 v5.0.1 h1:LX8T5B9FN8KZXOtxgN+R3I4THRRVB6+28IKgKBpXmAM=
556556
github.com/ory/analytics-go/v5 v5.0.1/go.mod h1:lWCiCjAaJkKfgR/BN5DCLMol8BjKS1x+4jxBxff/FF0=
557-
github.com/ory/client-go v1.22.41 h1:AywohwpZUMDVwnPaAoeZWseapSEF58GgPn1RPUvTaqQ=
558-
github.com/ory/client-go v1.22.41/go.mod h1:G1f+5+m/PJVvl40bsRn0QuyVIcXe7EHiWeM7iWpIDjw=
557+
github.com/ory/client-go v1.22.51 h1:T5tmhDvomkPTZeHQgfDcqjRnSvV1wkkG2xYQ/r6TQdk=
558+
github.com/ory/client-go v1.22.51/go.mod h1:G1f+5+m/PJVvl40bsRn0QuyVIcXe7EHiWeM7iWpIDjw=
559559
github.com/ory/dockertest/v4 v4.0.0 h1:i19aFsO/VXE0VrMk4ifnKW4G/KIJ93PCjLOslxXoPME=
560560
github.com/ory/dockertest/v4 v4.0.0/go.mod h1:b5Ofu8VIxWNhXFvQcLu17pRNQdoUBKtXBW74G4Ygzx8=
561561
github.com/ory/go-acc v0.2.9-0.20230103102148-6b1c9a70dbbe h1:rvu4obdvqR0fkSIJ8IfgzKOWwZ5kOT2UNfLq81Qk7rc=

0 commit comments

Comments
 (0)