query alerts

cmd/noaalert/main.go

@@ -50,6 +50,29 @@ func main() {
 			Usage:    "get active NOAA alerts",
 			Action:   alerts,
 		},
+		{
+			Name:     "query",
+			Category: "utility",
+			Usage:    "query the event stream for previous events",
+			Action:   query,
+			Flags: []cli.Flag{
+				&cli.IntFlag{
+					Name:    "offset",
+					Aliases: []string{"O"},
+					Value:   0,
+				},
+				&cli.IntFlag{
+					Name:    "limit",
+					Aliases: []string{"l"},
+					Value:   20,
+				},
+				&cli.StringFlag{
+					Name:    "output",
+					Aliases: []string{"o"},
+					Value:   "query.jsonlines",
+				},
+			},
+		},
 		{
 			Name:     "config",
 			Usage:    "print noaalerts configuration guide",
@@ -181,3 +204,39 @@ func projectInfo(c *cli.Context) (err error) {
 
 	return nil
 }
+
+func query(c *cli.Context) (err error) {
+	var conf noaalert.Config
+	if conf, err = noaalert.NewConfig(); err != nil {
+		return cli.Exit(err, 1)
+	}
+
+	var client *ensign.Client
+	if client, err = ensign.New(conf.Ensign.Options()...); err != nil {
+		return cli.Exit(err, 1)
+	}
+
+	var alerts *noaalert.AlertIterator
+	if alerts, err = noaalert.Query(client, c.Int("offset"), c.Int("limit")); err != nil {
+		return cli.Exit(err, 1)
+	}
+	defer alerts.Release()
+
+	var f *os.File
+	if f, err = os.Create(c.String("output")); err != nil {
+		return cli.Exit(err, 1)
+	}
+	defer f.Close()
+
+	for alerts.Next() {
+		if _, err = f.Write(alerts.Alert().Data); err != nil {
+			return cli.Exit(err, 1)
+		}
+		f.Write([]byte("\n"))
+	}
+
+	if err = alerts.Error(); err != nil {
+		return cli.Exit(err, 1)
+	}
+	return nil
+}

go.mod

@@ -5,7 +5,7 @@ go 1.20
 require (
 	github.com/joho/godotenv v1.5.1
 	github.com/rotationalio/confire v1.0.0
-	github.com/rotationalio/go-ensign v0.9.0
+	github.com/rotationalio/go-ensign v0.9.1
 	github.com/rs/zerolog v1.30.0
 	github.com/stretchr/testify v1.8.4
 	github.com/urfave/cli/v2 v2.25.7

go.sum

@@ -30,8 +30,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rotationalio/confire v1.0.0 h1:Ex1jtwVyvuMhFY0EXfgbMsvd9MPO5V9LvJZ0q740M9k=
 github.com/rotationalio/confire v1.0.0/go.mod h1:ug7pBDiZZl/4JjXJ2Effmj+L+0T2DBbG+Us1qQcRex0=
-github.com/rotationalio/go-ensign v0.9.0 h1:oGQoncRR0Jo1d07D0K8lq4MrsisVRIXIbFCz1L1aYSA=
-github.com/rotationalio/go-ensign v0.9.0/go.mod h1://SkPpSs3SLoeUt2NAzaUtGOpIccbnPoNc64MJpu9Dc=
+github.com/rotationalio/go-ensign v0.9.1 h1:jGyYLk+35RlP2Vylj+CQbBiYT3yZud+50x9JUjEeBZM=
+github.com/rotationalio/go-ensign v0.9.1/go.mod h1://SkPpSs3SLoeUt2NAzaUtGOpIccbnPoNc64MJpu9Dc=
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.30.0 h1:SymVODrcRsaRaSInD9yQtKbtWqwsfoPcRff/oRXLj4c=
 github.com/rs/zerolog v1.30.0/go.mod h1:/tk+P47gFdPXq4QYjvCmT5/Gsug2nagsFWBWhAiSi1w=

query.go

@@ -0,0 +1,70 @@
+package noaalert
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/rotationalio/go-ensign"
+	api "github.com/rotationalio/go-ensign/api/v1beta1"
+)
+
+func Query(client *ensign.Client, offset, limit int) (_ *AlertIterator, err error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+
+	var cursor *ensign.QueryCursor
+	if cursor, err = client.EnSQL(ctx, &api.Query{Query: fmt.Sprintf("SELECT * FROM noaa-alerts OFFSET %d LIMIT %d", offset, limit)}); err != nil {
+		return nil, err
+	}
+
+	return &AlertIterator{cursor: cursor}, nil
+}
+
+type AlertIterator struct {
+	err     error
+	done    bool
+	current *AlertEvent
+	cursor  *ensign.QueryCursor
+}
+
+func (a *AlertIterator) Next() bool {
+	if a.done {
+		return false
+	}
+
+	event, err := a.cursor.FetchOne()
+	if err != nil {
+		if !errors.Is(err, ensign.ErrNoRows) {
+			a.err = err
+		}
+		a.done = true
+		return false
+	}
+
+	a.current = &AlertEvent{
+		CorrelationID: event.Metadata["correlation_id"],
+		RequestID:     event.Metadata["request_id"],
+		ServerID:      event.Metadata["server_id"],
+		LastModified:  event.Metadata["last_modified"],
+		Expires:       event.Metadata["expires"],
+		Data:          event.Data,
+	}
+
+	return true
+}
+
+func (a *AlertIterator) Alert() *AlertEvent {
+	return a.current
+}
+
+func (a *AlertIterator) Error() error {
+	return a.err
+}
+
+func (a *AlertIterator) Release() {
+	if err := a.cursor.Close(); err != nil {
+		a.err = errors.Join(a.err, err)
+	}
+}