this repo has no description
refactor: rework shutdowns
+35
-32
+27
-22
cmd/mostliked/main.go
+27
-22
cmd/mostliked/main.go
···
6
6
"os"
7
7
"os/signal"
8
8
"syscall"
9
+
"context"
9
10
10
11
"github.com/edavis/bsky-feeds/pkg/mostliked"
11
12
"github.com/gorilla/websocket"
12
13
_ "github.com/mattn/go-sqlite3"
13
14
)
15
+
16
+
//const JetstreamUrl = `wss://jetstream1.us-west.bsky.network/subscribe?wantedCollections=app.bsky.feed.post&wantedCollections=app.bsky.feed.like&cursor=1728846514000000`
17
+
const JetstreamUrl = `ws://localhost:6008/subscribe?wantedCollections=app.bsky.feed.post&wantedCollections=app.bsky.feed.like`
14
18
15
19
func main() {
16
-
conn, _, err := websocket.DefaultDialer.Dial("ws://localhost:6008/subscribe?wantedCollections=app.bsky.feed.post&wantedCollections=app.bsky.feed.like", nil)
20
+
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
21
+
defer stop()
22
+
23
+
conn, _, err := websocket.DefaultDialer.Dial(JetstreamUrl, nil)
17
24
if err != nil {
18
-
log.Fatal("websocket connection error:", err)
25
+
log.Fatalf("failed to open websocket: %v\n", err)
19
26
}
20
-
defer conn.Close()
27
+
defer func() {
28
+
if err := conn.Close(); err != nil {
29
+
log.Printf("failed to close websocket: %v\n", err)
30
+
}
31
+
log.Println("websocket closed")
32
+
}()
21
33
22
34
dbCnx, err := sql.Open("sqlite3", "data/mostliked.db?_journal=WAL&_fk=on")
23
35
if err != nil {
24
-
log.Fatal("error opening db")
36
+
log.Fatalf("failed to open database: %v\n", err)
25
37
}
26
-
defer dbCnx.Close()
38
+
defer func() {
39
+
if err := dbCnx.Close(); err != nil {
40
+
log.Printf("failed to close db: %v", err)
41
+
}
42
+
log.Println("db closed")
43
+
}()
27
44
28
45
jetstreamEvents := make(chan []byte)
29
-
go mostliked.Handler(jetstreamEvents, dbCnx)
30
-
31
-
signalChan := make(chan os.Signal, 1)
32
-
cleanupDone := make(chan struct{})
33
-
signal.Notify(signalChan, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
34
-
35
-
go func(conn *websocket.Conn, dbCnx *sql.DB, jetstreamEvents chan []byte) {
36
-
<-signalChan
37
-
log.Println("shutting down...")
38
-
conn.Close()
39
-
dbCnx.Close()
40
-
close(cleanupDone)
41
-
}(conn, dbCnx, jetstreamEvents)
46
+
go mostliked.Handler(ctx, jetstreamEvents, dbCnx)
42
47
43
48
log.Println("starting up")
44
-
go func(conn *websocket.Conn) {
49
+
go func() {
45
50
for {
46
51
_, message, err := conn.ReadMessage()
47
52
if err != nil {
48
-
log.Println("error reading from websocket:", err)
49
53
break
50
54
}
51
55
jetstreamEvents <- message
52
56
}
53
-
}(conn)
57
+
}()
54
58
55
-
<-cleanupDone
59
+
<-ctx.Done()
60
+
log.Println("shutting down")
56
61
}
+1
-1
pkg/mostliked/generator.go
+1
-1
pkg/mostliked/generator.go
+7
-9
pkg/mostliked/handler.go
+7
-9
pkg/mostliked/handler.go
···
62
62
return ""
63
63
}
64
64
65
-
func Handler(events <-chan []byte, dbCnx *sql.DB) {
66
-
ctx := context.Background()
67
-
65
+
func Handler(ctx context.Context, events <-chan []byte, dbCnx *sql.DB) {
68
66
if _, err := dbCnx.ExecContext(ctx, ddl); err != nil {
69
-
log.Fatal("couldn't create tables")
67
+
log.Printf("couldn't create tables: %v\n", err)
70
68
}
71
69
queries := db.New(dbCnx)
72
70
···
100
98
if !txOpen {
101
99
dbTx, err = dbCnx.BeginTx(ctx, nil)
102
100
if err != nil {
103
-
log.Fatal(err)
101
+
log.Printf("failed to begin transaction: %v\n", err)
104
102
}
105
103
txOpen = true
106
104
queriesTx = queries.WithTx(dbTx)
···
158
156
continue
159
157
}
160
158
drafts.Delete(like.Subject.Uri)
161
-
log.Println("storing", like.Subject.Uri, "in database")
159
+
// log.Println("storing", like.Subject.Uri, "in database")
162
160
err := queriesTx.InsertPost(ctx, db.InsertPostParams{
163
161
Uri: like.Subject.Uri,
164
162
CreateTs: draftPost.Created,
···
184
182
}
185
183
186
184
eventCount += 1
187
-
if eventCount % 500 == 0 {
185
+
if eventCount % 1000 == 0 {
188
186
if err := dbTx.Commit(); err != nil {
189
-
log.Fatalf("commit failed: %v\n", err)
187
+
log.Printf("commit failed: %v\n", err)
190
188
} else {
191
189
txOpen = false
192
190
}
193
-
log.Println("db committed")
191
+
// log.Println("db committed")
194
192
}
195
193
}
196
194
}