this repo has no description
add the files
+294
+105
cmd/feed-generator/main.go
+105
cmd/feed-generator/main.go
···
1
+
package main
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"fmt"
7
+
"log"
8
+
"log/slog"
9
+
"os"
10
+
"os/signal"
11
+
"path"
12
+
"syscall"
13
+
14
+
"tangled.sh/willdot.net/feed-demo-go"
15
+
16
+
"github.com/avast/retry-go/v4"
17
+
"github.com/joho/godotenv"
18
+
)
19
+
20
+
const (
21
+
defaultJetstreamAddr = "wss://jetstream.atproto.tools/subscribe"
22
+
serverPort = 443 // this must be the port value used. See https://docs.bsky.app/docs/starter-templates/custom-feeds#deploying-your-feed
23
+
)
24
+
25
+
func main() {
26
+
err := run()
27
+
if err != nil {
28
+
log.Fatal(err)
29
+
}
30
+
}
31
+
32
+
func run() error {
33
+
err := godotenv.Load()
34
+
if err != nil && !os.IsNotExist(err) {
35
+
return fmt.Errorf("error loading .env file")
36
+
}
37
+
38
+
signals := make(chan os.Signal, 1)
39
+
signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
40
+
41
+
feedHost := os.Getenv("FEED_HOST_NAME")
42
+
if feedHost == "" {
43
+
return fmt.Errorf("FEED_HOST_NAME not set")
44
+
}
45
+
feedName := os.Getenv("FEED_NAME")
46
+
if feedName == "" {
47
+
return fmt.Errorf("FEED_NAME not set")
48
+
}
49
+
50
+
dbPath := os.Getenv("DATABASE_PATH")
51
+
if dbPath == "" {
52
+
dbPath = "./"
53
+
}
54
+
55
+
dbFilename := path.Join(dbPath, "database.db")
56
+
database, err := feed.NewDatabase(dbFilename)
57
+
if err != nil {
58
+
return fmt.Errorf("create new store: %w", err)
59
+
}
60
+
defer database.Close()
61
+
62
+
ctx, cancel := context.WithCancel(context.Background())
63
+
defer cancel()
64
+
65
+
go consumeLoop(ctx, database)
66
+
67
+
server, err := feed.NewServer(serverPort, feedHost, feedName, database)
68
+
if err != nil {
69
+
return fmt.Errorf("create new server: %w", err)
70
+
}
71
+
go func() {
72
+
<-signals
73
+
cancel()
74
+
_ = server.Stop(context.Background())
75
+
}()
76
+
77
+
server.Run()
78
+
return nil
79
+
}
80
+
81
+
func consumeLoop(ctx context.Context, database *feed.Database) {
82
+
handler := feed.NewFeedHandler(database)
83
+
84
+
jsServerAddr := os.Getenv("JS_SERVER_ADDR")
85
+
if jsServerAddr == "" {
86
+
jsServerAddr = defaultJetstreamAddr
87
+
}
88
+
89
+
consumer := feed.NewJetstreamConsumer(jsServerAddr, slog.Default(), handler)
90
+
91
+
_ = retry.Do(func() error {
92
+
err := consumer.Consume(ctx)
93
+
if err != nil {
94
+
// if the context has been cancelled then it's time to exit
95
+
if errors.Is(err, context.Canceled) {
96
+
return nil
97
+
}
98
+
slog.Error("consume loop", "error", err)
99
+
return err
100
+
}
101
+
return nil
102
+
}, retry.Attempts(0)) // retry indefinitly until context canceled
103
+
104
+
slog.Warn("exiting consume loop")
105
+
}
+189
cmd/register-feed/main.go
+189
cmd/register-feed/main.go
···
1
+
package main
2
+
3
+
import (
4
+
"bytes"
5
+
"encoding/json"
6
+
"fmt"
7
+
"io"
8
+
"log/slog"
9
+
"net/http"
10
+
"os"
11
+
"time"
12
+
13
+
"github.com/joho/godotenv"
14
+
)
15
+
16
+
const (
17
+
baseurl = "https://bsky.social/xrpc"
18
+
19
+
httpClientTimeoutDuration = time.Second * 5
20
+
transportIdleConnTimeoutDuration = time.Second * 90
21
+
)
22
+
23
+
type auth struct {
24
+
AccessJwt string `json:"accessJwt"`
25
+
Did string `json:"did"`
26
+
}
27
+
28
+
type registerFeedGen struct {
29
+
Repo string `json:"repo"`
30
+
Collection string `json:"collection"`
31
+
Rkey string `json:"rkey"`
32
+
Record registerRecord `json:"record"`
33
+
}
34
+
35
+
type registerRecord struct {
36
+
Did string `json:"did"`
37
+
DisplayName string `json:"displayName"`
38
+
Description string `json:"description"`
39
+
CreatedAt time.Time `json:"createdAt"`
40
+
}
41
+
42
+
func main() {
43
+
err := run()
44
+
if err != nil {
45
+
slog.Error("error registering feed", "error", err)
46
+
os.Exit(1)
47
+
}
48
+
}
49
+
50
+
func run() error {
51
+
err := godotenv.Load()
52
+
if err != nil && !os.IsNotExist(err) {
53
+
return fmt.Errorf("error loading .env file")
54
+
}
55
+
56
+
httpClient := http.Client{
57
+
Timeout: httpClientTimeoutDuration,
58
+
Transport: &http.Transport{
59
+
IdleConnTimeout: transportIdleConnTimeoutDuration,
60
+
},
61
+
}
62
+
auth, err := login(httpClient)
63
+
if err != nil {
64
+
return fmt.Errorf("failed to login: %w", err)
65
+
}
66
+
67
+
err = Register(auth, httpClient)
68
+
if err != nil {
69
+
return err
70
+
}
71
+
72
+
return nil
73
+
}
74
+
75
+
func login(client http.Client) (*auth, error) {
76
+
handle := os.Getenv("BSKY_HANDLE")
77
+
appPass := os.Getenv("BSKY_PASS")
78
+
79
+
url := fmt.Sprintf("%s/com.atproto.server.createsession", baseurl)
80
+
81
+
requestData := map[string]interface{}{
82
+
"identifier": handle,
83
+
"password": appPass,
84
+
}
85
+
86
+
data, err := json.Marshal(requestData)
87
+
if err != nil {
88
+
return nil, fmt.Errorf("failed to marshal request: %w", err)
89
+
}
90
+
91
+
r := bytes.NewReader(data)
92
+
93
+
req, err := http.NewRequest("POST", url, r)
94
+
if err != nil {
95
+
return nil, fmt.Errorf("failed to create request: %w", err)
96
+
}
97
+
98
+
req.Header.Add("Content-Type", "application/json")
99
+
100
+
res, err := client.Do(req)
101
+
if err != nil {
102
+
return nil, fmt.Errorf("failed to make request: %w", err)
103
+
}
104
+
105
+
defer func() {
106
+
_ = res.Body.Close()
107
+
}()
108
+
109
+
resBody, err := io.ReadAll(res.Body)
110
+
if err != nil {
111
+
return nil, fmt.Errorf("failed to read response: %w", err)
112
+
}
113
+
114
+
var loginResp auth
115
+
err = json.Unmarshal(resBody, &loginResp)
116
+
if err != nil {
117
+
return nil, fmt.Errorf("failed to unmarshal response: %w", err)
118
+
}
119
+
return &loginResp, nil
120
+
}
121
+
122
+
func Register(auth *auth, httpClient http.Client) error {
123
+
feedName := os.Getenv("FEED_NAME")
124
+
if feedName == "" {
125
+
return fmt.Errorf("FEED_NAME env not set")
126
+
}
127
+
feedDisplayName := os.Getenv("FEED_DISPLAY_NAME")
128
+
if feedDisplayName == "" {
129
+
return fmt.Errorf("FEED_DISPLAY_NAME env not set")
130
+
}
131
+
feedDescription := os.Getenv("FEED_DESCRIPTION")
132
+
if feedDescription == "" {
133
+
return fmt.Errorf("FEED_DESCRIPTION env not set")
134
+
}
135
+
feedDID := os.Getenv("FEED_DID")
136
+
if feedDID == "" {
137
+
return fmt.Errorf("FEED_DID environment not set")
138
+
}
139
+
140
+
reqData := registerFeedGen{
141
+
Repo: auth.Did,
142
+
Collection: "app.bsky.feed.generator",
143
+
Rkey: feedName,
144
+
Record: registerRecord{
145
+
Did: feedDID,
146
+
DisplayName: feedDisplayName,
147
+
Description: feedDescription,
148
+
CreatedAt: time.Now(),
149
+
},
150
+
}
151
+
152
+
data, err := json.Marshal(reqData)
153
+
if err != nil {
154
+
return fmt.Errorf("failed to marshal request: %w", err)
155
+
}
156
+
157
+
r := bytes.NewReader(data)
158
+
159
+
url := fmt.Sprintf("%s/com.atproto.repo.putRecord", baseurl)
160
+
req, err := http.NewRequest("POST", url, r)
161
+
if err != nil {
162
+
return fmt.Errorf("failed to create new post request: %w", err)
163
+
}
164
+
165
+
req.Header.Add("Content-Type", "application/json")
166
+
req.Header.Add("Authorization", fmt.Sprintf("Bearer %s", auth.AccessJwt))
167
+
168
+
res, err := httpClient.Do(req)
169
+
if err != nil {
170
+
return fmt.Errorf("failed to make create post request: %w", err)
171
+
}
172
+
173
+
defer func() {
174
+
_ = res.Body.Close()
175
+
}()
176
+
177
+
b, err := io.ReadAll(res.Body)
178
+
if err != nil {
179
+
fmt.Println(err)
180
+
} else {
181
+
fmt.Println(string(b))
182
+
}
183
+
184
+
if res.StatusCode != 200 {
185
+
return fmt.Errorf("failed to create post: %v", res.StatusCode)
186
+
}
187
+
188
+
return nil
189
+
}