whey.party
A collection of Custom Bluesky Feeds, including Fresh Feeds, all under one roof
redo
+7
.env.example
+7
.env.example
+6
-1
.gitignore
+6
-1
.gitignore
+63
-49
README.md
+63
-49
README.md
···
1
-
# go-bsky-feed-generator
2
-
A minimal implementation of a BlueSky Feed Generator in Go
3
-
4
-
5
-
## Requirements
1
+
# Rinds
2
+
A collection of feeds under one roof.
6
3
7
-
To run this feed generator, all you need is `docker` with `docker-compose`.
4
+
I don't like Docker and I don't need to compile this, okay thanks!
8
5
9
6
## Running
10
-
11
-
Start up the feed generator by running: `make up`
12
-
13
-
This will build the feed generator service binary inside a docker container and stand up the service on your machine at port `9032`.
14
-
15
-
To view a sample static feed (with only one post) go to:
16
-
17
-
- [`http://localhost:9032/xrpc/app.bsky.feed.getFeedSkeleton?feed=at://did:plc:replace-me-with-your-did/app.bsky.feed.generator/static`](http://localhost:9032/xrpc/app.bsky.feed.getFeedSkeleton?feed=at://did:plc:replace-me-with-your-did/app.bsky.feed.generator/static)
18
-
19
-
Update the variables in `.env` when you actually want to deploy the service somewhere, at which point `did:plc:replace-me-with-your-did` should be replaced with the value of `FEED_ACTOR_DID`.
20
-
21
-
## Accessing
22
-
23
-
This service exposes the following routes:
24
-
25
-
- `/.well-known/did.json`
26
-
- This route is used by ATProto to verify ownership of the DID the service is claiming, it's a static JSON document.
27
-
- You can see how this is generated in `pkg/gin/endpoints.go:GetWellKnownDID()`
28
-
- `/xrpc/app.bsky.feed.getFeedSkeleton`
29
-
- This route is what clients call to generate a feed page, it includes three query parameters for feed generation: `feed`, `cursor`, and `limit`
30
-
- You can see how those are parsed and handled in `pkg/gin/endpoints.go:GetFeedSkeleton()`
31
-
- `/xrpc/app.bsky.feed.describeFeedGenerator`
32
-
- This route is how the service advertises which feeds it supports to clients.
33
-
- You can see how those are parsed and handled in `pkg/gin/endpoints.go:DescribeFeeds()`
34
-
35
-
## Publishing
36
-
37
-
Once you've got your feed generator up and running and have it exposed to the internet, you can publish the feed using the script from the official BSky repo [here](https://github.com/bluesky-social/feed-generator/blob/main/scripts/publishFeedGen.ts).
7
+
aint that hard, go install go, setup postgres, and generate the required feed definitions under your user account. you can use https://pdsls.dev to generate a `app.bsky.feed.generator` record.
8
+
Set the `rkey` to the desired short url value. itll look like:
9
+
```
10
+
https://bsky.app/profile/{you}/feed/{rkey}
11
+
```
12
+
for the contents you can use the example below
13
+
```
14
+
{
15
+
"did": "did:web:${INSERT DID:WEB HERE}",
16
+
"$type": "app.bsky.feed.generator",
17
+
"createdAt": "2025-01-21T11:33:02.396Z",
18
+
"description": "wowww very descriptive",
19
+
"displayName": "Cool Feed Name",
20
+
}
21
+
```
38
22
39
-
Your feed will be published under _your_ DID and should show up in your profile under the `feeds` tab.
23
+
## Env
24
+
You can check out `.env.example` for an example
40
25
41
-
## Architecture
42
26
43
-
This repo is structured to abstract away a `Feed` interface that allows for you to add all sorts of feeds to the router.
27
+
## Postgres
28
+
Be sure to set up `.env` correctly
44
29
45
-
These feeds can be simple static feeds like the `pkg/feeds/static/feed.go` implementation, or they can be much more complex feeds that draw on different data sources and filter them in cool ways to produce pages of feed items.
30
+
All relevant tables should be created automatically when needed.
46
31
47
-
The `Feed` interface is defined by any struct implementing two functions:
32
+
## Index
33
+
You should start Postgres first
34
+
Then go run the firehose ingester in
35
+
```
36
+
cd ./indexer
37
+
```
38
+
and go compile it
39
+
```
40
+
go build -o indexer ./indexer.go && export $(grep -v '^#' ./../.env | xargs) && ./indexer
41
+
```
42
+
after it has been compiled, you can use `rerun.sh` to ensure it will automatically recover after failure
48
43
49
-
``` go
50
-
type Feed interface {
51
-
GetPage(ctx context.Context, feed string, userDID string, limit int64, cursor string) (feedPosts []*appbsky.FeedDefs_SkeletonFeedPost, newCursor *string, err error)
52
-
Describe(ctx context.Context) ([]appbsky.FeedDescribeFeedGenerator_Feed, error)
53
-
}
44
+
## Serve
45
+
Make sure the indexer (or at least Postgres) is running first:
46
+
```
47
+
go build -o feedgen cmd/main.go && export $(grep -v '^#' ./.env | xargs) && ./feedgen
54
48
```
49
+
the logs are pretty verbose imo, fyi
55
50
56
-
`GetPage` gets a page of a feed for a given user with the limit and cursor provided, this is the main function that serves posts to a user.
51
+
## Todo
52
+
- [ ] Faster Indexing
53
+
- [ ] Proper Up-to-Date Following Indexing
54
+
- [x] Repost Indicators
55
+
- [ ] Cache Timeouts
56
+
- [x] Likes
57
+
- [x] Posts
58
+
- [x] Feed Caches
59
+
- [ ] Followings
60
+
- [ ] More Fresh Feed Variants
61
+
- [ ] unFresh
62
+
- [x] +9 hrs
63
+
- [ ] Glimpse
64
+
- [ ] Media
65
+
- [ ] Fresh: Gram
66
+
- [ ] Fresh: Tube
67
+
- [ ] Fresh: Media Only
68
+
- [ ] Fresh: Text Only
57
69
58
-
`Describe` is used by the router to advertise what feeds are available, for foward compatibility, `Feed`s should be self describing in case this endpoint allows more details about feeds to be provided.
70
+
## Architecture
71
+
Based on [go-bsky-feed-generator](https://github.com/ericvolp12/go-bsky-feed-generator). Read the README in the linked repo for more info about how it all works.
59
72
60
-
You can configure external resources and requirements in your Feed implementation before `Adding` the feed to the `FeedRouter` with `feedRouter.AddFeed([]string{"{feed_name}"}, feedInstance)`
73
+
### /feeds/static
74
+
Basic example feed from the template. Kept as a sanity check if all else seems to fail.
61
75
62
-
This `Feed` interface is somewhat flexible right now but it could be better. I'm not sure if it will change in the future so keep that in mind when using this template.
76
+
### /feeds/fresh
77
+
Fresh feeds, all based around a shared Following feed builder and logic to set posts as viewed. May contain some remnant old references to the old name "rinds".
63
78
64
-
- This has since been updated to allow a Feed to take in a feed name when generating a page and register multiple aliases for feeds that are supported.
+129
cmd/main.go
+129
cmd/main.go
···
2
2
3
3
import (
4
4
"context"
5
+
"database/sql"
5
6
"fmt"
6
7
"log"
7
8
"net/http"
8
9
"net/url"
9
10
"os"
10
11
"time"
12
+
13
+
_ "github.com/lib/pq"
11
14
12
15
auth "github.com/ericvolp12/go-bsky-feed-generator/pkg/auth"
13
16
"github.com/ericvolp12/go-bsky-feed-generator/pkg/feedrouter"
14
17
ginendpoints "github.com/ericvolp12/go-bsky-feed-generator/pkg/gin"
15
18
19
+
freshfeeds "github.com/ericvolp12/go-bsky-feed-generator/pkg/feeds/fresh"
16
20
staticfeed "github.com/ericvolp12/go-bsky-feed-generator/pkg/feeds/static"
17
21
ginprometheus "github.com/ericvolp12/go-gin-prometheus"
18
22
"github.com/gin-gonic/gin"
···
27
31
28
32
func main() {
29
33
ctx := context.Background()
34
+
35
+
// Open the database connection
36
+
dbHost := os.Getenv("DB_HOST")
37
+
dbUser := os.Getenv("DB_USER")
38
+
dbName := os.Getenv("DB_NAME")
39
+
dbPassword := os.Getenv("DB_PASSWORD")
40
+
db, err := sql.Open("postgres", fmt.Sprintf("user=%s dbname=%s host=%s password=%s sslmode=disable", dbUser, dbName, dbHost, dbPassword))
41
+
if err != nil {
42
+
log.Fatalf("Failed to open database: %v", err)
43
+
}
44
+
defer db.Close()
45
+
46
+
// Ping the database to ensure the connection is established
47
+
if err := db.Ping(); err != nil {
48
+
log.Fatalf("Failed to ping database: %v", err)
49
+
}
30
50
31
51
// Configure feed generator from environment variables
32
52
···
88
108
[]string{"at://did:plc:q6gjnaw2blty4crticxkmujt/app.bsky.feed.post/3jx7msc4ive26"},
89
109
)
90
110
111
+
// idk help me
112
+
113
+
rindsFeed, rindsFeedAliases, err := freshfeeds.NewStaticFeed(
114
+
ctx,
115
+
feedActorDID,
116
+
"rinds",
117
+
// This static post is the conversation that sparked this demo repo
118
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
119
+
db,
120
+
"rinds",
121
+
false,
122
+
)
123
+
124
+
randomFeed, randomFeedAliases, err := freshfeeds.NewStaticFeed(
125
+
ctx,
126
+
feedActorDID,
127
+
"random",
128
+
// This static post is the conversation that sparked this demo repo
129
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
130
+
db,
131
+
"random",
132
+
false,
133
+
)
134
+
135
+
repostsFeed, repostsFeedAliases, err := freshfeeds.NewStaticFeed(
136
+
ctx,
137
+
feedActorDID,
138
+
"reposts",
139
+
// This static post is the conversation that sparked this demo repo
140
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
141
+
db,
142
+
"reposts",
143
+
false,
144
+
)
145
+
mnineFeed, mnineFeedAliases, err := freshfeeds.NewStaticFeed(
146
+
ctx,
147
+
feedActorDID,
148
+
"mnine",
149
+
// This static post is the conversation that sparked this demo repo
150
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
151
+
db,
152
+
"mnine",
153
+
false,
154
+
)
155
+
156
+
rrindsFeed, rrindsFeedAliases, err := freshfeeds.NewStaticFeed(
157
+
ctx,
158
+
feedActorDID,
159
+
"rinds-replies",
160
+
// This static post is the conversation that sparked this demo repo
161
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
162
+
db,
163
+
"rinds",
164
+
true,
165
+
)
166
+
167
+
rrandomFeed, rrandomFeedAliases, err := freshfeeds.NewStaticFeed(
168
+
ctx,
169
+
feedActorDID,
170
+
"random-replies",
171
+
// This static post is the conversation that sparked this demo repo
172
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
173
+
db,
174
+
"random",
175
+
true,
176
+
)
177
+
178
+
rrepostsFeed, rrepostsFeedAliases, err := freshfeeds.NewStaticFeed(
179
+
ctx,
180
+
feedActorDID,
181
+
"reposts-replies",
182
+
// This static post is the conversation that sparked this demo repo
183
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
184
+
db,
185
+
"reposts",
186
+
true,
187
+
)
188
+
rmnineFeed, rmnineFeedAliases, err := freshfeeds.NewStaticFeed(
189
+
ctx,
190
+
feedActorDID,
191
+
"mnine-replies",
192
+
// This static post is the conversation that sparked this demo repo
193
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
194
+
db,
195
+
"mnine",
196
+
true,
197
+
)
198
+
orepliesFeed, orepliesFeedAliases, err := freshfeeds.NewStaticFeed(
199
+
ctx,
200
+
feedActorDID,
201
+
"oreplies",
202
+
// This static post is the conversation that sparked this demo repo
203
+
[]string{"at://did:plc:mn45tewwnse5btfftvd3powc/app.bsky.feed.post/3kgjjhlsnoi2f"},
204
+
db,
205
+
"oreplies",
206
+
true,
207
+
)
91
208
// Add the static feed to the feed generator
92
209
feedRouter.AddFeed(staticFeedAliases, staticFeed)
210
+
211
+
feedRouter.AddFeed(rindsFeedAliases, rindsFeed)
212
+
feedRouter.AddFeed(randomFeedAliases, randomFeed)
213
+
feedRouter.AddFeed(repostsFeedAliases, repostsFeed)
214
+
feedRouter.AddFeed(mnineFeedAliases, mnineFeed)
215
+
216
+
feedRouter.AddFeed(rrindsFeedAliases, rrindsFeed)
217
+
feedRouter.AddFeed(rrandomFeedAliases, rrandomFeed)
218
+
feedRouter.AddFeed(rrepostsFeedAliases, rrepostsFeed)
219
+
feedRouter.AddFeed(rmnineFeedAliases, rmnineFeed)
220
+
221
+
feedRouter.AddFeed(orepliesFeedAliases, orepliesFeed)
93
222
94
223
// Create a gin router with default middleware for logging and recovery
95
224
router := gin.Default()
+1
go.mod
+1
go.mod
···
70
70
github.com/lestrrat-go/iter v1.0.2 // indirect
71
71
github.com/lestrrat-go/jwx/v2 v2.0.12 // indirect
72
72
github.com/lestrrat-go/option v1.0.1 // indirect
73
+
github.com/lib/pq v1.10.9
73
74
github.com/mattn/go-isatty v0.0.20 // indirect
74
75
github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
75
76
github.com/minio/sha256-simd v1.0.1 // indirect
+2
go.sum
+2
go.sum
···
142
142
github.com/lestrrat-go/option v1.0.0/go.mod h1:5ZHFbivi4xwXxhxY9XHDe2FHo6/Z7WWmtT7T5nBBp3I=
143
143
github.com/lestrrat-go/option v1.0.1 h1:oAzP2fvZGQKWkvHa1/SAcFolBEca1oN+mQ7eooNBEYU=
144
144
github.com/lestrrat-go/option v1.0.1/go.mod h1:5ZHFbivi4xwXxhxY9XHDe2FHo6/Z7WWmtT7T5nBBp3I=
145
+
github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
146
+
github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
145
147
github.com/mattn/go-isatty v0.0.14/go.mod h1:7GGIvUiUoEMVVmxf/4nioHXj79iQHKdU27kJ6hsGG94=
146
148
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
147
149
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+10
indexer/go.mod
+10
indexer/go.mod
+8
indexer/go.sum
+8
indexer/go.sum
···
1
+
github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw=
2
+
github.com/go-chi/chi/v5 v5.1.0/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8=
3
+
github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
4
+
github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
5
+
github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
6
+
github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
7
+
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
8
+
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
+474
indexer/indexer.go
+474
indexer/indexer.go
···
1
+
package main
2
+
3
+
import (
4
+
"context"
5
+
"database/sql"
6
+
"fmt"
7
+
"log"
8
+
"os"
9
+
"time"
10
+
11
+
"github.com/gorilla/websocket"
12
+
"github.com/lib/pq"
13
+
_ "github.com/lib/pq"
14
+
)
15
+
16
+
const wsUrl = "wss://jetstream2.us-west.bsky.network/subscribe?wantedCollections=app.bsky.feed.post&wantedCollections=app.bsky.feed.repost&wantedCollections=app.bsky.feed.like"
17
+
18
+
type LikeMessage struct {
19
+
Did string `json:"did"`
20
+
TimeUs int64 `json:"time_us"`
21
+
Kind string `json:"kind"`
22
+
Commit Commit `json:"commit"`
23
+
}
24
+
25
+
type Commit struct {
26
+
Rev string `json:"rev"`
27
+
Operation string `json:"operation"`
28
+
Collection string `json:"collection"`
29
+
RKey string `json:"rkey"`
30
+
Record LikeRecord `json:"record"`
31
+
CID string `json:"cid"`
32
+
}
33
+
34
+
type LikeRecord struct {
35
+
Type string `json:"$type"`
36
+
CreatedAt string `json:"createdAt"`
37
+
Subject LikeSubject `json:"subject"`
38
+
Reply *Reply `json:"reply,omitempty"`
39
+
}
40
+
41
+
type Reply struct {
42
+
Parent ReplySubject `json:"parent"`
43
+
Root ReplySubject `json:"root"`
44
+
}
45
+
46
+
type LikeSubject struct {
47
+
CID string `json:"cid"`
48
+
URI string `json:"uri"`
49
+
}
50
+
51
+
type ReplySubject struct {
52
+
CID string `json:"cid"`
53
+
URI string `json:"uri"`
54
+
}
55
+
56
+
var lastLoggedSecond int64 // Keep track of the last logged second
57
+
58
+
var (
59
+
postBatch []Post
60
+
likeBatch []Like
61
+
batchInsertSize = 1000 // Adjust the batch size as needed
62
+
batchInterval = 30 * time.Second // Flush every 30 seconds
63
+
)
64
+
65
+
type Post struct {
66
+
RelAuthor string
67
+
PostUri string
68
+
RelDate int64
69
+
IsRepost bool
70
+
RepostUri string
71
+
ReplyTo string
72
+
}
73
+
74
+
type Like struct {
75
+
RelAuthor string
76
+
PostUri string
77
+
RelDate int64
78
+
}
79
+
80
+
func getLastCursor(db *sql.DB) int64 {
81
+
var lastCursor int64
82
+
err := db.QueryRow("SELECT lastCursor FROM cursor WHERE id = 1").Scan(&lastCursor)
83
+
if err != nil {
84
+
if err == sql.ErrNoRows {
85
+
log.Println("Cursor table is empty; starting fresh.")
86
+
return 0
87
+
}
88
+
log.Fatalf("Error fetching last cursor: %v", err)
89
+
}
90
+
return lastCursor
91
+
}
92
+
93
+
func main() {
94
+
// Connect to Postgres // Open the database connection
95
+
dbHost := os.Getenv("DB_HOST")
96
+
dbUser := os.Getenv("DB_USER")
97
+
dbName := os.Getenv("DB_NAME")
98
+
dbPassword := os.Getenv("DB_PASSWORD")
99
+
db, err := sql.Open("postgres", fmt.Sprintf("user=%s dbname=%s host=%s password=%s sslmode=disable", dbUser, dbName, dbHost, dbPassword))
100
+
101
+
if err != nil {
102
+
log.Fatalf("Failed to connect to Postgres: %v", err)
103
+
}
104
+
defer db.Close()
105
+
106
+
// Ensure tables exist
107
+
createTables(db)
108
+
109
+
// Start the cleanup job
110
+
go startCleanupJob(db)
111
+
112
+
// Start the batch insert job
113
+
go startBatchInsertJob(db)
114
+
115
+
// Start the batch insert job for likes
116
+
go startBatchInsertLikesJob(db)
117
+
118
+
// Retrieve the last cursor
119
+
lastCursor := getLastCursor(db)
120
+
121
+
// If the cursor is older than 24 hours, skip it
122
+
if lastCursor > 0 {
123
+
cursorTime := time.UnixMicro(lastCursor)
124
+
if time.Since(cursorTime) > 24*time.Hour {
125
+
log.Printf("Cursor is older than 24 hours (%s); skipping it.", cursorTime.Format("2006-01-02 15:04:05"))
126
+
lastCursor = 0 // Ignore this cursor
127
+
} else {
128
+
log.Printf("Resuming from cursor: %d (%s)", lastCursor, cursorTime.Format("2006-01-02 15:04:05"))
129
+
}
130
+
}
131
+
132
+
// WebSocket URL with cursor if available
133
+
wsFullUrl := wsUrl
134
+
if lastCursor > 0 {
135
+
wsFullUrl += "&cursor=" + fmt.Sprintf("%d", lastCursor)
136
+
}
137
+
138
+
// Connect to WebSocket
139
+
conn, _, err := websocket.DefaultDialer.Dial(wsFullUrl, nil)
140
+
if err != nil {
141
+
log.Fatalf("WebSocket connection error: %v", err)
142
+
}
143
+
defer conn.Close()
144
+
145
+
//print wsFullUrl
146
+
log.Printf("Connected to WebSocket: %s", wsFullUrl)
147
+
148
+
log.Println("Listening for WebSocket messages...")
149
+
150
+
// Process WebSocket messages
151
+
for {
152
+
var msg LikeMessage
153
+
err := conn.ReadJSON(&msg)
154
+
if err != nil {
155
+
log.Printf("Error reading WebSocket message: %v", err)
156
+
continue
157
+
}
158
+
159
+
processMessage(db, msg)
160
+
}
161
+
}
162
+
163
+
func createTables(db *sql.DB) {
164
+
_, err := db.Exec(`
165
+
CREATE TABLE IF NOT EXISTS posts (
166
+
id SERIAL PRIMARY KEY,
167
+
rel_author TEXT NOT NULL,
168
+
post_uri TEXT NOT NULL,
169
+
rel_date BIGINT NOT NULL,
170
+
is_repost BOOLEAN NOT NULL DEFAULT FALSE,
171
+
repost_uri TEXT,
172
+
reply_to TEXT,
173
+
UNIQUE(rel_author, post_uri, rel_date)
174
+
);
175
+
`)
176
+
if err != nil {
177
+
log.Fatalf("Error creating 'posts' table: %v", err)
178
+
}
179
+
180
+
_, err = db.Exec(`
181
+
CREATE TABLE IF NOT EXISTS likes (
182
+
id SERIAL PRIMARY KEY,
183
+
rel_author TEXT NOT NULL,
184
+
post_uri TEXT NOT NULL,
185
+
rel_date BIGINT NOT NULL
186
+
);
187
+
`)
188
+
if err != nil {
189
+
log.Fatalf("Error creating 'posts' table: %v", err)
190
+
}
191
+
192
+
// Create a cursor table with a single-row constraint
193
+
_, err = db.Exec(`
194
+
CREATE TABLE IF NOT EXISTS cursor (
195
+
id INT PRIMARY KEY CHECK (id = 1),
196
+
lastCursor BIGINT NOT NULL
197
+
);
198
+
`)
199
+
if err != nil {
200
+
log.Fatalf("Error creating 'cursor' table: %v", err)
201
+
}
202
+
203
+
// Ensure the cursor table always has exactly one row
204
+
_, err = db.Exec(`
205
+
INSERT INTO cursor (id, lastCursor)
206
+
VALUES (1, 0)
207
+
ON CONFLICT (id) DO NOTHING;
208
+
`)
209
+
if err != nil {
210
+
log.Fatalf("Error initializing cursor table: %v", err)
211
+
}
212
+
}
213
+
func processMessage(db *sql.DB, msg LikeMessage) {
214
+
// Convert cursor to time
215
+
cursorTime := time.UnixMicro(msg.TimeUs)
216
+
217
+
// Get the whole second as a Unix timestamp
218
+
currentSecond := cursorTime.Unix()
219
+
220
+
// Check if this second has already been logged
221
+
if currentSecond != lastLoggedSecond && cursorTime.Nanosecond() >= 100_000_000 && cursorTime.Nanosecond() < 200_000_000 {
222
+
// Update the last logged second
223
+
lastLoggedSecond = currentSecond
224
+
225
+
// Log only once per second
226
+
humanReadableTime := cursorTime.Format("2006-01-02 15:04:05.000")
227
+
log.Printf("Cursor (time_us): %d, Human-readable time: %s", msg.TimeUs, humanReadableTime)
228
+
}
229
+
230
+
// Save the record
231
+
record := msg.Commit.Record
232
+
postUri := fmt.Sprintf("at://%s/app.bsky.feed.post/%s", msg.Did, msg.Commit.RKey)
233
+
repostUri := fmt.Sprintf("at://%s/app.bsky.feed.repost/%s", msg.Did, msg.Commit.RKey)
234
+
reply := ""
235
+
if msg.Commit.Record.Reply != nil {
236
+
reply = fmt.Sprintf("Parent: %s, Root: %s", msg.Commit.Record.Reply.Parent.URI, msg.Commit.Record.Reply.Root.URI)
237
+
}
238
+
239
+
switch msg.Commit.Collection {
240
+
case "app.bsky.feed.post":
241
+
if msg.Commit.Operation == "create" {
242
+
postBatch = append(postBatch, Post{msg.Did, postUri, msg.TimeUs, false, "", reply})
243
+
} else if msg.Commit.Operation == "delete" {
244
+
deletePost(db, msg.Did, postUri, msg.TimeUs)
245
+
}
246
+
case "app.bsky.feed.repost":
247
+
if record.Subject.URI != "" {
248
+
if msg.Commit.Operation == "create" {
249
+
postBatch = append(postBatch, Post{msg.Did, record.Subject.URI, msg.TimeUs, true, repostUri, ""})
250
+
} else if msg.Commit.Operation == "delete" {
251
+
deletePost(db, msg.Did, record.Subject.URI, msg.TimeUs)
252
+
}
253
+
}
254
+
case "app.bsky.feed.like":
255
+
if record.Subject.URI != "" {
256
+
if msg.Commit.Operation == "create" {
257
+
likeBatch = append(likeBatch, Like{msg.Did, record.Subject.URI, msg.TimeUs})
258
+
} else if msg.Commit.Operation == "delete" {
259
+
deleteLike(db, msg.Did, record.Subject.URI)
260
+
}
261
+
}
262
+
default:
263
+
//log.Printf("Unknown collection: %s", msg.Commit.Collection)
264
+
}
265
+
266
+
// Update the cursor in the single-row table
267
+
_, err := db.Exec(`
268
+
UPDATE cursor SET lastCursor = $1 WHERE id = 1;
269
+
`, msg.TimeUs)
270
+
if err != nil {
271
+
log.Printf("Error updating cursor: %v", err)
272
+
}
273
+
}
274
+
275
+
func deletePost(db *sql.DB, relAuthor, postUri string, relDate int64) {
276
+
_, err := db.Exec(`
277
+
DELETE FROM posts WHERE rel_author = $1 AND post_uri = $2 AND rel_date = $3;
278
+
`, relAuthor, postUri, relDate)
279
+
if err != nil {
280
+
log.Printf("Error deleting post: %v", err)
281
+
}
282
+
}
283
+
284
+
func deleteLike(db *sql.DB, relAuthor, postUri string) {
285
+
_, err := db.Exec(`
286
+
DELETE FROM likes WHERE rel_author = $1 AND post_uri = $2;
287
+
`, relAuthor, postUri)
288
+
if err != nil {
289
+
log.Printf("Error deleting like: %v", err)
290
+
}
291
+
}
292
+
293
+
func startCleanupJob(db *sql.DB) {
294
+
ticker := time.NewTicker(1 * time.Hour)
295
+
defer ticker.Stop()
296
+
297
+
for range ticker.C {
298
+
cleanupOldPosts(db)
299
+
if err := cleanOldFeedCaches(context.Background(), db); err != nil {
300
+
log.Printf("Error cleaning old feed caches: %v\n", err)
301
+
}
302
+
}
303
+
}
304
+
305
+
func cleanupOldPosts(db *sql.DB) {
306
+
threshold := time.Now().Add(-24 * time.Hour).UnixMicro()
307
+
_, err := db.Exec(`
308
+
DELETE FROM posts WHERE rel_date < $1;
309
+
`, threshold)
310
+
if err != nil {
311
+
log.Printf("Error deleting old posts: %v", err)
312
+
} else {
313
+
log.Printf("Deleted posts older than 24 hours.")
314
+
}
315
+
}
316
+
317
+
func startBatchInsertJob(db *sql.DB) {
318
+
ticker := time.NewTicker(batchInterval)
319
+
defer ticker.Stop()
320
+
321
+
for range ticker.C {
322
+
if len(postBatch) >= batchInsertSize {
323
+
batchInsertPosts(db)
324
+
}
325
+
}
326
+
}
327
+
328
+
func batchInsertPosts(db *sql.DB) {
329
+
tx, err := db.Begin()
330
+
if err != nil {
331
+
log.Printf("Error starting transaction: %v", err)
332
+
return
333
+
}
334
+
335
+
stmt, err := tx.Prepare(`
336
+
INSERT INTO posts (rel_author, post_uri, rel_date, is_repost, repost_uri, reply_to)
337
+
VALUES ($1, $2, $3, $4, $5, $6)
338
+
ON CONFLICT (rel_author, post_uri, rel_date) DO NOTHING;
339
+
`)
340
+
if err != nil {
341
+
log.Printf("Error preparing statement: %v", err)
342
+
return
343
+
}
344
+
defer stmt.Close()
345
+
346
+
for _, post := range postBatch {
347
+
_, err := stmt.Exec(post.RelAuthor, post.PostUri, post.RelDate, post.IsRepost, post.RepostUri, post.ReplyTo)
348
+
if err != nil {
349
+
log.Printf("Error executing statement: %v", err)
350
+
}
351
+
}
352
+
353
+
err = tx.Commit()
354
+
if err != nil {
355
+
log.Printf("Error committing transaction: %v", err)
356
+
}
357
+
358
+
// Clear the batch
359
+
postBatch = postBatch[:0]
360
+
}
361
+
362
+
func startBatchInsertLikesJob(db *sql.DB) {
363
+
ticker := time.NewTicker(1 * time.Second)
364
+
defer ticker.Stop()
365
+
366
+
for range ticker.C {
367
+
if len(likeBatch) > 0 {
368
+
batchInsertLikes(db)
369
+
}
370
+
}
371
+
}
372
+
373
+
func batchInsertLikes(db *sql.DB) {
374
+
tx, err := db.Begin()
375
+
if err != nil {
376
+
log.Printf("Error starting transaction: %v", err)
377
+
return
378
+
}
379
+
380
+
stmt, err := tx.Prepare(`
381
+
INSERT INTO likes (rel_author, post_uri, rel_date)
382
+
VALUES ($1, $2, $3)
383
+
ON CONFLICT (rel_author, post_uri) DO NOTHING;
384
+
`)
385
+
if err != nil {
386
+
log.Printf("Error preparing statement: %v", err)
387
+
return
388
+
}
389
+
defer stmt.Close()
390
+
391
+
for _, like := range likeBatch {
392
+
_, err := stmt.Exec(like.RelAuthor, like.PostUri, like.RelDate)
393
+
if err != nil {
394
+
log.Printf("Error executing statement: %v", err)
395
+
}
396
+
}
397
+
398
+
err = tx.Commit()
399
+
if err != nil {
400
+
log.Printf("Error committing transaction: %v", err)
401
+
}
402
+
403
+
// Clear the batch
404
+
likeBatch = likeBatch[:0]
405
+
}
406
+
407
+
func cleanOldFeedCaches(ctx context.Context, db *sql.DB) error {
408
+
//log
409
+
log.Println("Cleaning old feed caches")
410
+
// Get the current time minus 24 hours
411
+
expirationTime := time.Now().Add(-24 * time.Hour)
412
+
413
+
// Get all tables from cachetimeout that are older than 24 hours
414
+
rows, err := db.QueryContext(ctx, `
415
+
SELECT table_name
416
+
FROM cachetimeout
417
+
WHERE creation_time < $1
418
+
`, expirationTime)
419
+
if err != nil {
420
+
return fmt.Errorf("error querying cachetimeout table: %w", err)
421
+
}
422
+
defer rows.Close()
423
+
424
+
var tablesToDelete []string
425
+
for rows.Next() {
426
+
var tableName string
427
+
if err := rows.Scan(&tableName); err != nil {
428
+
return fmt.Errorf("error scanning table name: %w", err)
429
+
}
430
+
tablesToDelete = append(tablesToDelete, tableName)
431
+
}
432
+
433
+
if err := rows.Err(); err != nil {
434
+
return fmt.Errorf("error iterating cachetimeout rows: %w", err)
435
+
}
436
+
437
+
// Get all feedcache_* tables that do not have an entry in cachetimeout
438
+
rows, err = db.QueryContext(ctx, `
439
+
SELECT table_name
440
+
FROM information_schema.tables
441
+
WHERE table_name LIKE 'feedcache_%'
442
+
AND table_name NOT IN (SELECT table_name FROM cachetimeout)
443
+
`)
444
+
if err != nil {
445
+
return fmt.Errorf("error querying feedcache tables: %w", err)
446
+
}
447
+
defer rows.Close()
448
+
449
+
for rows.Next() {
450
+
var tableName string
451
+
if err := rows.Scan(&tableName); err != nil {
452
+
return fmt.Errorf("error scanning table name: %w", err)
453
+
}
454
+
tablesToDelete = append(tablesToDelete, tableName)
455
+
}
456
+
457
+
if err := rows.Err(); err != nil {
458
+
return fmt.Errorf("error iterating feedcache rows: %w", err)
459
+
}
460
+
461
+
// Drop the old tables and remove their entries from cachetimeout
462
+
for _, tableName := range tablesToDelete {
463
+
_, err := db.ExecContext(ctx, fmt.Sprintf("DROP TABLE IF EXISTS %s", pq.QuoteIdentifier(tableName)))
464
+
if err != nil {
465
+
return fmt.Errorf("error dropping table %s: %w", tableName, err)
466
+
}
467
+
_, err = db.ExecContext(ctx, "DELETE FROM cachetimeout WHERE table_name = $1", tableName)
468
+
if err != nil {
469
+
return fmt.Errorf("error deleting from cachetimeout table: %w", err)
470
+
}
471
+
}
472
+
473
+
return nil
474
+
}
+11
indexer/rerun.sh
+11
indexer/rerun.sh
+2
-1
pkg/auth/auth.go
+2
-1
pkg/auth/auth.go
···
115
115
accessToken := authHeaderParts[1]
116
116
117
117
parser := jwt.Parser{
118
-
ValidMethods: []string{es256k.SigningMethodES256K.Alg()},
118
+
ValidMethods: []string{es256k.SigningMethodES256K.Alg()},
119
+
SkipClaimsValidation: true, // IM SORRY I HAD TO, MY VPS IS ACTING STRANGE. I THINK ITS FINE
119
120
}
120
121
121
122
token, err := parser.ParseWithClaims(accessToken, claims, func(token *jwt.Token) (interface{}, error) {
+571
pkg/feeds/fresh/feed.go
+571
pkg/feeds/fresh/feed.go
···
1
+
package rinds
2
+
3
+
import (
4
+
"context"
5
+
"crypto/sha256"
6
+
"database/sql"
7
+
"encoding/hex"
8
+
"encoding/json"
9
+
"fmt"
10
+
"io/ioutil"
11
+
"log"
12
+
"sort"
13
+
"strconv"
14
+
"strings"
15
+
"time"
16
+
17
+
"net/http"
18
+
19
+
"math/rand"
20
+
21
+
appbsky "github.com/bluesky-social/indigo/api/bsky"
22
+
"github.com/lib/pq"
23
+
)
24
+
25
+
type CachedFollows struct {
26
+
Follows []string `json:"follows"`
27
+
LastModified time.Time `json:"last_modified"`
28
+
}
29
+
30
+
type FeedType string
31
+
32
+
const (
33
+
Rinds FeedType = "rinds"
34
+
Random FeedType = "random"
35
+
Mnine FeedType = "mnine"
36
+
Reposts FeedType = "reposts"
37
+
OReplies FeedType = "oreplies"
38
+
)
39
+
40
+
type StaticFeed struct {
41
+
FeedActorDID string
42
+
FeedName string
43
+
StaticPostURIs []string
44
+
DB *sql.DB
45
+
FeedType FeedType // random, mnine, reposts
46
+
RepliesOn bool
47
+
}
48
+
49
+
type Follower struct {
50
+
DID string `json:"did"`
51
+
}
52
+
53
+
type Response struct {
54
+
Follows []Follower `json:"follows"`
55
+
Cursor string `json:"cursor"`
56
+
}
57
+
58
+
type PostWithDate struct {
59
+
PostURI string `json:"post_uri"`
60
+
RelDate int64 `json:"rel_date"`
61
+
IsRepost bool `json:"is_repost"`
62
+
RepostURI string `json:"repost_uri,omitempty"`
63
+
}
64
+
65
+
type CachedPosts struct {
66
+
Posts []PostWithDate `json:"posts"`
67
+
LastModified time.Time `json:"last_modified"`
68
+
}
69
+
70
+
// Describe implements feedrouter.Feed.
71
+
func (sf *StaticFeed) Describe(ctx context.Context) ([]appbsky.FeedDescribeFeedGenerator_Feed, error) {
72
+
panic("unimplemented")
73
+
}
74
+
75
+
// NewStaticFeed returns a new StaticFeed, a list of aliases for the feed, and an error
76
+
// StaticFeed is a trivial implementation of the Feed interface, so its aliases are just the input feedName
77
+
func NewStaticFeed(ctx context.Context, feedActorDID string, feedName string, staticPostURIs []string, db *sql.DB, feedType FeedType, repliesOn bool) (*StaticFeed, []string, error) {
78
+
return &StaticFeed{
79
+
FeedActorDID: feedActorDID,
80
+
FeedName: feedName,
81
+
StaticPostURIs: staticPostURIs,
82
+
DB: db,
83
+
FeedType: feedType,
84
+
RepliesOn: repliesOn,
85
+
}, []string{feedName}, nil
86
+
}
87
+
88
+
// GetPage returns a list of FeedDefs_SkeletonFeedPost, a new cursor, and an error
89
+
// It takes a feed name, a user DID, a limit, and a cursor
90
+
// The feed name can be used to produce different feeds from the same feed generator
91
+
func (sf *StaticFeed) GetPage(ctx context.Context, feed string, userDID string, limit int64, cursor string) ([]*appbsky.FeedDefs_SkeletonFeedPost, *string, error) {
92
+
cursorAsInt := int64(0)
93
+
var hash string
94
+
var startOfLastPage int64
95
+
var sizeOfLastPage int64
96
+
var inflightstartOfLastPage int64
97
+
var inflightsizeOfLastPage int64
98
+
var err error
99
+
100
+
var smartReadEnabled bool = true
101
+
var smartReportingEnabled bool = true
102
+
log.Printf("smartReadEnabled is %v; smartReportingEnabled is %v", smartReadEnabled, smartReportingEnabled)
103
+
104
+
if cursor != "" {
105
+
parts := strings.Split(cursor, "-")
106
+
if len(parts) != 6 {
107
+
return nil, nil, fmt.Errorf("invalid cursor format")
108
+
}
109
+
cursorAsInt, err = strconv.ParseInt(parts[0], 10, 64)
110
+
if err != nil {
111
+
return nil, nil, fmt.Errorf("cursor is not an integer: %w", err)
112
+
}
113
+
hash = parts[1]
114
+
inflightstartOfLastPage, err = strconv.ParseInt(parts[2], 10, 64)
115
+
if err != nil {
116
+
return nil, nil, fmt.Errorf("start of last page is not an integer: %w", err)
117
+
}
118
+
inflightsizeOfLastPage, err = strconv.ParseInt(parts[3], 10, 64)
119
+
if err != nil {
120
+
return nil, nil, fmt.Errorf("size of last page is not an integer: %w", err)
121
+
}
122
+
startOfLastPage, err = strconv.ParseInt(parts[4], 10, 64)
123
+
if err != nil {
124
+
return nil, nil, fmt.Errorf("start of last page is not an integer: %w", err)
125
+
}
126
+
sizeOfLastPage, err = strconv.ParseInt(parts[5], 10, 64)
127
+
if err != nil {
128
+
return nil, nil, fmt.Errorf("size of last page is not an integer: %w", err)
129
+
}
130
+
}
131
+
132
+
if limit == 1 && cursor == "" {
133
+
// this happens when the app tries to check if the timeline has new posts at the top or not
134
+
// we should handle this better but ehhhhhhhh
135
+
log.Print("limit is 1 and cursor is empty. Skipping the database query.")
136
+
return nil, nil, nil
137
+
} else if cursor == "" {
138
+
log.Println("Generating new hash")
139
+
hash = generateHash(userDID)
140
+
} else {
141
+
log.Println("Using existing hash")
142
+
}
143
+
144
+
tableName := fmt.Sprintf("feedcache_%s_%s", userDID, hash)
145
+
146
+
// Check if cache table exists
147
+
var exists bool
148
+
err = sf.DB.QueryRowContext(ctx, "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = $1)", tableName).Scan(&exists)
149
+
if err != nil {
150
+
return nil, nil, fmt.Errorf("error checking cache table existence: %w", err)
151
+
}
152
+
153
+
if exists {
154
+
rows, err := sf.DB.QueryContext(ctx, fmt.Sprintf("SELECT post_uri, rel_date, is_repost, repost_uri FROM %s", pq.QuoteIdentifier(tableName)))
155
+
if err != nil {
156
+
return nil, nil, fmt.Errorf("error querying cache table: %w", err)
157
+
}
158
+
defer rows.Close()
159
+
160
+
var cachedPosts []PostWithDate
161
+
for rows.Next() {
162
+
var post PostWithDate
163
+
var repostURI sql.NullString
164
+
if err := rows.Scan(&post.PostURI, &post.RelDate, &post.IsRepost, &repostURI); err != nil {
165
+
return nil, nil, fmt.Errorf("error scanning cached post: %w", err)
166
+
}
167
+
if repostURI.Valid {
168
+
post.RepostURI = repostURI.String
169
+
}
170
+
cachedPosts = append(cachedPosts, post)
171
+
}
172
+
173
+
if err := rows.Err(); err != nil {
174
+
return nil, nil, fmt.Errorf("error iterating cached posts: %w", err)
175
+
}
176
+
177
+
log.Printf("Cached posts found: %d", len(cachedPosts))
178
+
// fuck it print the entrie cachedPosts
179
+
180
+
// Mark posts from the last page as viewed
181
+
/*
182
+
// Handle empty cache table
183
+
if len(cachedPosts) == 0 || int64(len(cachedPosts)) < cursorAsInt {
184
+
// just return https://bsky.app/profile/strandingy.bsky.social/post/3lgfdgn7i6s2q
185
+
// special case when you reach the end
186
+
markPostsAsViewed(ctx, sf.DB, userDID, cachedPosts, smartReportingEnabled)
187
+
emptysinglepost := []PostWithDate{}
188
+
emptysinglepost = append(emptysinglepost, PostWithDate{
189
+
PostURI: "at://did:plc:css3l47v2r4xhcgykfd5mdmn/app.bsky.feed.post/3lgfdgn7i6s2q",
190
+
RelDate: time.Now().UnixNano() / int64(time.Microsecond),
191
+
})
192
+
log.Print("empty page hanglerer paginateResult for cachedPosts")
193
+
return paginateResults(emptysinglepost, inflightstartOfLastPage, inflightsizeOfLastPage, cursorAsInt, limit, hash)
194
+
}
195
+
*/
196
+
197
+
if cursor != "" {
198
+
lastPagePosts := cachedPosts[startOfLastPage : startOfLastPage+sizeOfLastPage]
199
+
// the normal intended markPostsAsViewed call
200
+
if err := markPostsAsViewed(ctx, sf.DB, userDID, lastPagePosts, smartReportingEnabled); err != nil {
201
+
return nil, nil, fmt.Errorf("error marking posts as viewed: %w", err)
202
+
}
203
+
log.Printf("Marking these posts as viewed. Range: %d - %d", startOfLastPage, startOfLastPage+sizeOfLastPage)
204
+
}
205
+
206
+
log.Print("default paginateResult for cachedPosts")
207
+
return paginateResults(cachedPosts, inflightstartOfLastPage, inflightsizeOfLastPage, cursorAsInt, limit, hash)
208
+
}
209
+
210
+
posts := []PostWithDate{}
211
+
212
+
log.Println("Fetching followers")
213
+
followers, err := getFollowers(ctx, sf.DB, userDID)
214
+
if err != nil {
215
+
log.Printf("Error fetching followers: %v\n", err)
216
+
return nil, nil, err
217
+
}
218
+
//log.Printf("Raw response: %s\n", followers)
219
+
220
+
log.Println("Converting followers to comma-separated string")
221
+
followerIDs := make([]string, len(followers))
222
+
copy(followerIDs, followers)
223
+
224
+
if len(followerIDs) == 0 {
225
+
log.Println("No followers found. Skipping the database query.")
226
+
return nil, nil, nil
227
+
}
228
+
229
+
//log.Println("Follower IDs:", followerIDs)
230
+
// Check if the table exists
231
+
var tableExists bool
232
+
err = sf.DB.QueryRowContext(ctx, fmt.Sprintf(`
233
+
SELECT EXISTS (
234
+
SELECT FROM information_schema.tables
235
+
WHERE table_name = %s
236
+
)`, pq.QuoteLiteral(fmt.Sprintf("viewedby_%s", userDID)))).Scan(&tableExists)
237
+
if err != nil {
238
+
log.Printf("Error checking table existence: %v\n", err)
239
+
return nil, nil, fmt.Errorf("error checking table existence: %w", err)
240
+
}
241
+
242
+
query := `
243
+
SELECT post_uri, rel_date, is_repost, repost_uri
244
+
FROM posts
245
+
WHERE rel_author = ANY($1)`
246
+
247
+
if smartReadEnabled {
248
+
query += `
249
+
AND post_uri NOT IN (SELECT post_uri FROM likes WHERE rel_author = $2)
250
+
AND post_uri NOT IN (SELECT post_uri FROM posts WHERE rel_author = $2 AND is_repost = TRUE)`
251
+
}
252
+
if !sf.RepliesOn {
253
+
query += " AND (reply_to IS NULL OR reply_to = '')"
254
+
}
255
+
if sf.FeedType == "reposts" {
256
+
query += " AND is_repost = TRUE"
257
+
}
258
+
if sf.FeedType == "oreplies" {
259
+
query += " AND (reply_to IS NOT NULL AND reply_to != '') AND is_repost = FALSE"
260
+
}
261
+
if tableExists && smartReadEnabled {
262
+
query += fmt.Sprintf(" AND post_uri NOT IN (SELECT post_uri FROM %s)", pq.QuoteIdentifier(fmt.Sprintf("viewedby_%s", userDID)))
263
+
}
264
+
var rows *sql.Rows
265
+
266
+
if smartReadEnabled {
267
+
if sf.FeedType == "mnine" {
268
+
query += ` AND rel_date < $3`
269
+
thresholdTime := time.Now().Add(-9*time.Hour).UnixNano() / int64(time.Microsecond)
270
+
rows, err = sf.DB.QueryContext(ctx, query, pq.Array(followerIDs), userDID, thresholdTime)
271
+
} else {
272
+
rows, err = sf.DB.QueryContext(ctx, query, pq.Array(followerIDs), userDID)
273
+
}
274
+
} else {
275
+
if sf.FeedType == "mnine" {
276
+
query += ` AND rel_date < $2`
277
+
thresholdTime := time.Now().Add(-9*time.Hour).UnixNano() / int64(time.Microsecond)
278
+
rows, err = sf.DB.QueryContext(ctx, query, pq.Array(followerIDs), thresholdTime)
279
+
} else {
280
+
rows, err = sf.DB.QueryContext(ctx, query, pq.Array(followerIDs))
281
+
}
282
+
}
283
+
log.Printf("Query: %s\n", query)
284
+
if err != nil {
285
+
log.Printf("Error querying posts: %v\n", err)
286
+
return nil, nil, fmt.Errorf("error querying posts: %w", err)
287
+
}
288
+
defer rows.Close()
289
+
290
+
log.Println("Iterating over rows")
291
+
for rows.Next() {
292
+
var postURI string
293
+
var relDate int64
294
+
var isRepost bool
295
+
var repostURI sql.NullString
296
+
if err := rows.Scan(&postURI, &relDate, &isRepost, &repostURI); err != nil {
297
+
log.Printf("error scanning post URI: %v\n", err)
298
+
return nil, nil, fmt.Errorf("error scanning post URI: %w", err)
299
+
}
300
+
post := PostWithDate{
301
+
PostURI: postURI,
302
+
RelDate: relDate,
303
+
IsRepost: isRepost,
304
+
}
305
+
if repostURI.Valid {
306
+
post.RepostURI = repostURI.String
307
+
}
308
+
posts = append(posts, post)
309
+
}
310
+
311
+
if err := rows.Err(); err != nil {
312
+
return nil, nil, fmt.Errorf("error iterating rows: %w", err)
313
+
}
314
+
315
+
log.Printf("Freshly queried posts found: %d", len(posts))
316
+
317
+
if sf.FeedType == "random" {
318
+
// Sort results randomly
319
+
log.Println("Sorting results randomly")
320
+
rand.Seed(time.Now().UnixNano())
321
+
rand.Shuffle(len(posts), func(i, j int) {
322
+
posts[i], posts[j] = posts[j], posts[i]
323
+
})
324
+
} else {
325
+
// Sort results by date
326
+
log.Println("Sorting results by date")
327
+
sort.Slice(posts, func(i, j int) bool {
328
+
return posts[i].RelDate > posts[j].RelDate
329
+
})
330
+
}
331
+
332
+
// Cache the results in the database
333
+
log.Println("Caching results in the database")
334
+
335
+
// Ensure the cachetimeout table exists
336
+
_, err = sf.DB.ExecContext(ctx, `
337
+
CREATE TABLE IF NOT EXISTS cachetimeout (
338
+
table_name TEXT UNIQUE,
339
+
creation_time TIMESTAMP
340
+
)
341
+
`)
342
+
if err != nil {
343
+
return nil, nil, fmt.Errorf("error creating cachetimeout table: %w", err)
344
+
}
345
+
346
+
_, err = sf.DB.ExecContext(ctx, fmt.Sprintf(`
347
+
CREATE TABLE %s (
348
+
post_uri TEXT,
349
+
rel_date BIGINT,
350
+
is_repost BOOLEAN,
351
+
repost_uri TEXT,
352
+
viewed BOOLEAN DEFAULT FALSE
353
+
)
354
+
`, pq.QuoteIdentifier(tableName)))
355
+
if err != nil {
356
+
return nil, nil, fmt.Errorf("error creating cache table: %w", err)
357
+
}
358
+
359
+
// Store the table name and creation time in cachetimeout
360
+
_, err = sf.DB.ExecContext(ctx, `
361
+
INSERT INTO cachetimeout (table_name, creation_time)
362
+
VALUES ($1, $2)
363
+
ON CONFLICT (table_name) DO NOTHING
364
+
`, tableName, time.Now())
365
+
if err != nil {
366
+
return nil, nil, fmt.Errorf("error inserting into cachetimeout table: %w", err)
367
+
}
368
+
369
+
for _, post := range posts {
370
+
_, err := sf.DB.ExecContext(ctx, fmt.Sprintf(`
371
+
INSERT INTO %s (post_uri, rel_date, is_repost, repost_uri)
372
+
VALUES ($1, $2, $3, $4)
373
+
`, pq.QuoteIdentifier(tableName)), post.PostURI, post.RelDate, post.IsRepost, post.RepostURI)
374
+
if err != nil {
375
+
return nil, nil, fmt.Errorf("error inserting into cache table: %w", err)
376
+
}
377
+
}
378
+
log.Print("default paginateResult for freshly queried posts")
379
+
return paginateResults(posts, inflightstartOfLastPage, inflightsizeOfLastPage, cursorAsInt, limit, hash)
380
+
}
381
+
382
+
func paginateResults(posts []PostWithDate, inflightcursorAsInt int64, inflightlimit int64, cursorAsInt int64, limit int64, hash string) ([]*appbsky.FeedDefs_SkeletonFeedPost, *string, error) {
383
+
log.Println("Paginating results")
384
+
var paginatedPosts []*appbsky.FeedDefs_SkeletonFeedPost
385
+
386
+
if int64(len(posts)) > cursorAsInt+limit {
387
+
for _, post := range posts[cursorAsInt : cursorAsInt+limit] {
388
+
paginatedPosts = append(paginatedPosts, formatPost(post))
389
+
}
390
+
newCursor := fmt.Sprintf("%d-%s-%d-%d-%d-%d", cursorAsInt+limit, hash, cursorAsInt, limit, inflightcursorAsInt, inflightlimit)
391
+
return paginatedPosts, &newCursor, nil
392
+
}
393
+
394
+
for _, post := range posts[cursorAsInt:] {
395
+
paginatedPosts = append(paginatedPosts, formatPost(post))
396
+
}
397
+
398
+
return paginatedPosts, nil, nil
399
+
}
400
+
401
+
func formatPost(post PostWithDate) *appbsky.FeedDefs_SkeletonFeedPost {
402
+
if post.IsRepost {
403
+
return &appbsky.FeedDefs_SkeletonFeedPost{
404
+
Post: post.PostURI,
405
+
Reason: &appbsky.FeedDefs_SkeletonFeedPost_Reason{
406
+
FeedDefs_SkeletonReasonRepost: &appbsky.FeedDefs_SkeletonReasonRepost{
407
+
Repost: post.RepostURI,
408
+
},
409
+
},
410
+
}
411
+
}
412
+
return &appbsky.FeedDefs_SkeletonFeedPost{
413
+
Post: post.PostURI,
414
+
}
415
+
}
416
+
417
+
func generateHash(input string) string {
418
+
hash := sha256.Sum256([]byte(input + time.Now().String()))
419
+
return hex.EncodeToString(hash[:])[:5]
420
+
}
421
+
422
+
func getFollowers(ctx context.Context, db *sql.DB, userdid string) ([]string, error) {
423
+
unquotedTableName := "follows_" + userdid
424
+
tableName := pq.QuoteIdentifier(unquotedTableName)
425
+
fmt.Printf("Checking for table: %s\n", tableName) // Debug log
426
+
427
+
// Check if cache table exists
428
+
var exists bool
429
+
query := "SELECT EXISTS (SELECT FROM information_schema.tables WHERE table_name = $1)"
430
+
fmt.Printf("Executing query: %s with parameter: %s\n", query, unquotedTableName) // Debug log
431
+
err := db.QueryRowContext(ctx, query, unquotedTableName).Scan(&exists)
432
+
if err != nil {
433
+
// Check for specific errors
434
+
if err == sql.ErrNoRows {
435
+
return nil, fmt.Errorf("table existence check returned no rows: %w", err)
436
+
}
437
+
if pqErr, ok := err.(*pq.Error); ok {
438
+
return nil, fmt.Errorf("PostgreSQL error: %s, Code: %s", pqErr.Message, pqErr.Code)
439
+
}
440
+
return nil, fmt.Errorf("error checking cache table existence: %w", err)
441
+
}
442
+
443
+
fmt.Printf("Table exists: %v\n", exists) // Debug log
444
+
445
+
if exists {
446
+
rows, err := db.QueryContext(ctx, fmt.Sprintf("SELECT follow FROM %s", tableName))
447
+
if err != nil {
448
+
return nil, fmt.Errorf("error querying cache table: %w", err)
449
+
}
450
+
defer rows.Close()
451
+
452
+
var cachedFollows []string
453
+
for rows.Next() {
454
+
var follow string
455
+
if err := rows.Scan(&follow); err != nil {
456
+
return nil, fmt.Errorf("error scanning cached follow: %w", err)
457
+
}
458
+
cachedFollows = append(cachedFollows, follow)
459
+
}
460
+
461
+
if err := rows.Err(); err != nil {
462
+
return nil, fmt.Errorf("error iterating cached follows: %w", err)
463
+
}
464
+
465
+
log.Println("Returning cached followers")
466
+
return cachedFollows, nil
467
+
}
468
+
469
+
log.Println("Fetching followers from API")
470
+
var allDIDs []string
471
+
cursor := ""
472
+
473
+
for {
474
+
apiURL := fmt.Sprintf("https://public.api.bsky.app/xrpc/app.bsky.graph.getFollows?actor=%s&cursor=%s", userdid, cursor)
475
+
resp, err := http.Get(apiURL)
476
+
if err != nil {
477
+
log.Printf("Error making request: %v\n", err)
478
+
return nil, fmt.Errorf("failed to make request: %v", err)
479
+
}
480
+
defer resp.Body.Close()
481
+
482
+
body, err := ioutil.ReadAll(resp.Body)
483
+
if err != nil {
484
+
log.Printf("Error reading response: %v\n", err)
485
+
return nil, fmt.Errorf("failed to read response: %v", err)
486
+
}
487
+
488
+
var response Response
489
+
if err := json.Unmarshal(body, &response); err != nil {
490
+
log.Printf("Error unmarshalling JSON: %v\n", err)
491
+
return nil, fmt.Errorf("failed to unmarshal JSON: %v", err)
492
+
}
493
+
494
+
for _, follow := range response.Follows {
495
+
allDIDs = append(allDIDs, follow.DID)
496
+
}
497
+
498
+
if response.Cursor == "" {
499
+
break
500
+
}
501
+
cursor = response.Cursor
502
+
}
503
+
504
+
// Drop the existing table if it exists
505
+
log.Println("Dropping existing followers table if it exists")
506
+
_, err = db.ExecContext(ctx, fmt.Sprintf("DROP TABLE IF EXISTS %s", tableName))
507
+
if err != nil {
508
+
return nil, fmt.Errorf("error dropping existing cache table: %w", err)
509
+
}
510
+
511
+
// Cache the results in the database
512
+
log.Println("Caching followers in the database")
513
+
_, err = db.ExecContext(ctx, fmt.Sprintf(`
514
+
CREATE TABLE %s (
515
+
follow TEXT UNIQUE
516
+
)
517
+
`, tableName))
518
+
if err != nil {
519
+
return nil, fmt.Errorf("error creating cache table: %w", err)
520
+
}
521
+
522
+
// Use a map to track unique follows
523
+
followMap := make(map[string]struct{})
524
+
for _, follow := range allDIDs {
525
+
if _, exists := followMap[follow]; !exists {
526
+
followMap[follow] = struct{}{}
527
+
_, err := db.ExecContext(ctx, fmt.Sprintf(`
528
+
INSERT INTO %s (follow)
529
+
VALUES ($1)
530
+
ON CONFLICT (follow) DO NOTHING
531
+
`, tableName), follow)
532
+
if err != nil {
533
+
return nil, fmt.Errorf("error inserting into cache table: %w", err)
534
+
}
535
+
}
536
+
}
537
+
538
+
log.Println("Returning fetched followers")
539
+
return allDIDs, nil
540
+
}
541
+
542
+
func markPostsAsViewed(ctx context.Context, db *sql.DB, userDID string, posts []PostWithDate, smartReportingEnabled bool) error {
543
+
if len(posts) == 0 || !smartReportingEnabled {
544
+
return nil
545
+
}
546
+
tableName := fmt.Sprintf("viewedby_%s", userDID)
547
+
548
+
// Create the table if it doesn't exist
549
+
_, err := db.ExecContext(ctx, fmt.Sprintf(`
550
+
CREATE TABLE IF NOT EXISTS %s (
551
+
post_uri TEXT UNIQUE
552
+
)
553
+
`, pq.QuoteIdentifier(tableName)))
554
+
if err != nil {
555
+
return fmt.Errorf("error creating viewed table: %w", err)
556
+
}
557
+
558
+
// Insert posts into the table
559
+
for _, post := range posts {
560
+
_, err := db.ExecContext(ctx, fmt.Sprintf(`
561
+
INSERT INTO %s (post_uri)
562
+
VALUES ($1)
563
+
ON CONFLICT (post_uri) DO NOTHING
564
+
`, pq.QuoteIdentifier(tableName)), post.PostURI)
565
+
if err != nil {
566
+
return fmt.Errorf("error inserting into viewed table: %w", err)
567
+
}
568
+
}
569
+
570
+
return nil
571
+
}