structure and shutdown

Author: 404salad

data/data.go

@@ -0,0 +1,36 @@
+package data 
+
+import (
+	"database/sql"
+	"log"
+
+	_ "modernc.org/sqlite"
+)
+
+func Init() *sql.DB {
+	db, err := sql.Open("sqlite", "posts.db")
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	_, err = db.Exec(`
+		CREATE TABLE IF NOT EXISTS posts (
+			id INTEGER PRIMARY KEY AUTOINCREMENT,
+			time_us INTEGER,
+			text TEXT
+		) STRICT
+	`)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	return db
+}
+
+func PrepareInsert(dbConn *sql.DB) *sql.Stmt {
+	stmt, err := dbConn.Prepare("INSERT INTO posts(time_us, text) VALUES (?, ?)")
+	if err != nil {
+		log.Fatal(err)
+	}
+	return stmt
+}

main.go

@@ -1,49 +1,22 @@
 package main
 
 import (
-	"database/sql"
+	"context"
 	"encoding/json"
+	"fmt"
 	"log"
 	"net/url"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
 
-	_ "modernc.org/sqlite"
+	"cave/data"
 	"github.com/gorilla/websocket"
+	_ "modernc.org/sqlite"
 )
 
-type MinimalEvent struct {
-	TimeUS int64 `json:"time_us"`
-	Commit struct {
-		Record struct {
-			Text string `json:"text"`
-		} `json:"record"`
-	} `json:"commit"`
-}
-
-func main() {
-
-	db, err := sql.Open("sqlite", "posts.db")
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer db.Close()
-
-	_, err = db.Exec(`
-		CREATE TABLE IF NOT EXISTS posts (
-			id INTEGER PRIMARY KEY AUTOINCREMENT,
-			time_us INTEGER,
-			text TEXT
-		) STRICT
-	`)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	stmt, err := db.Prepare("INSERT INTO posts(time_us, text) VALUES (?, ?)")
-	if err != nil {
-		log.Fatal(err)
-	}
-	defer stmt.Close()
-
+func websocket_dail() *websocket.Conn {
 	u := url.URL{
 		Scheme:   "wss",
 		Host:     "jetstream1.us-east.bsky.network",
@@ -55,26 +28,53 @@ func main() {
 	if err != nil {
 		log.Fatal("Dial error:", err)
 	}
+	return conn
+}
+
+func main() {
+	db := data.Init()
+	defer db.Close()
+
+	stmt := data.PrepareInsert(db)
+	defer stmt.Close()
+
+	conn := websocket_dail()
 	defer conn.Close()
 
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	log.Println("gathering data from the firehose...")
 	for {
-		_, message, err := conn.ReadMessage()
-		if err != nil {
-			log.Fatal("Read error:", err)
-		}
+		select {
+		case <-ctx.Done():
+			fmt.Print("\nwaiting for a proper shutdown")
+			for _ = range 3 {
+				time.Sleep(time.Second / 3)
+				fmt.Print(".")
+			}
+			fmt.Println("\nShutting down!")
+			return
+		default:
+			_, message, err := conn.ReadMessage()
+			if err != nil {
+				log.Fatal("Read error:", err)
+			}
 
-		var e MinimalEvent
-		if err := json.Unmarshal(message, &e); err != nil {
-			continue
-		}
+			var e MinimalEvent
+			if err := json.Unmarshal(message, &e); err != nil {
+				continue
+			}
 
-		if e.Commit.Record.Text == "" {
-			continue
-		}
+			if e.Commit.Record.Text == "" {
+				continue
+			}
 
-		_, err = stmt.Exec(e.TimeUS, e.Commit.Record.Text)
-		if err != nil {
-			log.Println("Insert error:", err)
+			_, err = stmt.Exec(e.TimeUS, e.Commit.Record.Text)
+			if err != nil {
+				log.Println("Insert error:", err)
+			}
 		}
+
 	}
 }

models.go

@@ -0,0 +1,10 @@
+package main
+
+type MinimalEvent struct {
+	TimeUS int64 `json:"time_us"`
+	Commit struct {
+		Record struct {
+			Text string `json:"text"`
+		} `json:"record"`
+	} `json:"commit"`
+}