this repo has no description
progress on basic indexing
+280
-13
+2
.gitignore
+2
.gitignore
+11
-2
README.md
+11
-2
README.md
···
1
1
2
-
scrumble
3
-
========
2
+
```
3
+
.::::::. .,-::::: :::::::.. ... :::. : :::::::. ::: .,::::::
4
+
;;;' ' ,;;;'''''' ;;;;'';;;; ;; ;;;;;,. ;;; ;;;'';;' ;;; ;;;;''''
5
+
'[==/[[[[,[[[ [[[,/[[[' [[' [[[[[[[, ,[[[[, [[[__[[\. [[[ [[cccc
6
+
''' $$$$ $$$$$$c $$ $$$$$$$$$$$"$$$ $$""""Y$$ $$' $$""""
7
+
88b dP'88bo,__,o, 888b "88bo,88 .d888888 Y88" 888o_88o,,od8Po88oo,.__888oo,__
8
+
"YMmMY" "YUMMMMMP"MMMM "W" "YmmMMMM""MMM M' "MMM""YUMMMP" """"YUMMM""""YUMMM
9
+
10
+
11
+
scrumble.social / make a scene
12
+
```
4
13
5
14
This is a work-in-progress atproto app for communities ("scenes") to currate and discuss arbitrary content.
+14
docker-compose.yml
+14
docker-compose.yml
···
1
+
version: "3.8"
2
+
services:
3
+
tap:
4
+
image: "ghcr.io/bluesky-social/indigo/tap:latest"
5
+
ports:
6
+
- "2480:2480"
7
+
volumes:
8
+
- type: bind
9
+
source: ./data/
10
+
target: /data
11
+
environment:
12
+
TAP_MAX_DB_CONNS: 12
13
+
TAP_SIGNAL_COLLECTION: social.scrumble.beta.account.profile
14
+
TAP_COLLECTION_FILTERS: social.scrumble.*
+5
go.mod
+5
go.mod
···
5
5
require (
6
6
github.com/bluesky-social/indigo v0.0.0-20251218205144-034a2c019e64
7
7
github.com/earthboundkid/versioninfo/v2 v2.24.1
8
+
github.com/gorilla/websocket v1.5.1
8
9
github.com/joho/godotenv v1.5.1
9
10
github.com/labstack/echo/v4 v4.11.3
10
11
github.com/prometheus/client_golang v1.17.0
12
+
github.com/stretchr/testify v1.11.1
11
13
github.com/urfave/cli/v3 v3.6.1
12
14
gorm.io/gorm v1.31.1
13
15
)
···
17
19
github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874 // indirect
18
20
github.com/cespare/xxhash/v2 v2.2.0 // indirect
19
21
github.com/cpuguy83/go-md2man/v2 v2.0.3 // indirect
22
+
github.com/davecgh/go-spew v1.1.1 // indirect
20
23
github.com/decred/dcrd/dcrec/secp256k1/v4 v4.2.0 // indirect
21
24
github.com/felixge/httpsnoop v1.0.4 // indirect
22
25
github.com/go-logr/logr v1.4.1 // indirect
···
70
73
github.com/multiformats/go-varint v0.0.7 // indirect
71
74
github.com/opentracing/opentracing-go v1.2.0 // indirect
72
75
github.com/orandin/slog-gorm v1.3.2 // indirect
76
+
github.com/pmezard/go-difflib v1.0.0 // indirect
73
77
github.com/polydawn/refmt v0.89.1-0.20221221234430-40501e09de1f // indirect
74
78
github.com/prometheus/client_model v0.5.0 // indirect
75
79
github.com/prometheus/common v0.45.0 // indirect
···
100
104
golang.org/x/time v0.3.0 // indirect
101
105
golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
102
106
google.golang.org/protobuf v1.33.0 // indirect
107
+
gopkg.in/yaml.v3 v3.0.1 // indirect
103
108
gorm.io/driver/postgres v1.5.7 // indirect
104
109
gorm.io/driver/sqlite v1.6.0 // indirect
105
110
lukechampine.com/blake3 v1.2.1 // indirect
+2
go.sum
+2
go.sum
···
40
40
github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
41
41
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
42
42
github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
43
+
github.com/gorilla/websocket v1.5.1 h1:gmztn0JnHVt9JZquRuzLw3g4wouNVzKL15iLr/zn/QY=
44
+
github.com/gorilla/websocket v1.5.1/go.mod h1:x3kM2JMyaluk02fnUJpQuwD2dCS5NDG2ZHL0uE0tcaY=
43
45
github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ=
44
46
github.com/hashicorp/go-cleanhttp v0.5.2/go.mod h1:kO/YDlP8L1346E6Sodw+PrpBSV4/SoxCXGY6BqNFT48=
45
47
github.com/hashicorp/go-hclog v0.9.2 h1:CG6TE5H9/JXsFWJCfoIVpKFIkFe6ysEuHirp4DxCsHI=
+181
indexer/indexer.go
+181
indexer/indexer.go
···
1
+
package indexer
2
+
3
+
import (
4
+
"context"
5
+
"encoding/json"
6
+
"log/slog"
7
+
8
+
"tangled.org/bnewbold.net/scrumble/api/socialscrumble"
9
+
"tangled.org/bnewbold.net/scrumble/store"
10
+
"tangled.org/bnewbold.net/scrumble/tapclient"
11
+
12
+
"github.com/bluesky-social/indigo/atproto/syntax"
13
+
)
14
+
15
+
type Indexer struct {
16
+
sceneDIDs []syntax.DID
17
+
store *store.Store
18
+
}
19
+
20
+
// if scene DIDs is empty, will index all scenes
21
+
func NewIndexer(st *store.Store, sceneDIDs []syntax.DID) (*Indexer, error) {
22
+
return &Indexer{
23
+
sceneDIDs: sceneDIDs,
24
+
store: st,
25
+
}, nil
26
+
}
27
+
28
+
func (idx *Indexer) HandleMessage(ctx context.Context, evt *tapclient.Event) error {
29
+
switch payload := evt.Payload().(type) {
30
+
case *tapclient.IdentityEvent:
31
+
return idx.HandleIdentity(ctx, payload)
32
+
case *tapclient.RecordEvent:
33
+
return idx.HandleRecord(ctx, payload)
34
+
}
35
+
return nil
36
+
}
37
+
38
+
func (idx *Indexer) HandleIdentity(ctx context.Context, evt *tapclient.IdentityEvent) error {
39
+
/*
40
+
type IdentityEvent struct {
41
+
DID string `json:"did"`
42
+
Handle string `json:"handle"`
43
+
IsActive bool `json:"isActive"`
44
+
Status string `json:"status"`
45
+
}
46
+
*/
47
+
slog.Info("received tap identity event", "did", evt.DID, "handle", evt.Handle)
48
+
did, err := syntax.ParseDID(evt.DID)
49
+
if err != nil {
50
+
return nil // skip
51
+
}
52
+
handle, err := syntax.ParseHandle(evt.Handle)
53
+
if err != nil {
54
+
handle = syntax.Handle("handle.invalid")
55
+
}
56
+
// TODO: validate Status / IsActive
57
+
if err := idx.store.UpsertIdentity(did, handle, evt.Status); err != nil {
58
+
return err
59
+
}
60
+
return nil
61
+
}
62
+
63
+
func (idx *Indexer) HandleRecord(ctx context.Context, evt *tapclient.RecordEvent) error {
64
+
/*
65
+
type RecordEvent struct {
66
+
DID string `json:"did"`
67
+
Collection string `json:"collection"`
68
+
Rkey string `json:"rkey"`
69
+
Action string `json:"action"`
70
+
CID string `json:"cid"`
71
+
Record json.RawMessage `json:"record"`
72
+
Live bool `json:"live"`
73
+
}
74
+
*/
75
+
logger := slog.With("did", evt.DID, "action", evt.Action, "collection", evt.Collection, "rkey", evt.Rkey)
76
+
logger.Info("received tap record event")
77
+
did, err := syntax.ParseDID(evt.DID)
78
+
if err != nil {
79
+
return nil // skip
80
+
}
81
+
collection, err := syntax.ParseNSID(evt.Collection)
82
+
if err != nil {
83
+
return nil // skip
84
+
}
85
+
rkey, err := syntax.ParseRecordKey(evt.Rkey)
86
+
if err != nil {
87
+
return nil // skip
88
+
}
89
+
90
+
switch evt.Collection {
91
+
case "social.scrumble.beta.account.profile":
92
+
if evt.Rkey != "self" {
93
+
return nil // skip
94
+
}
95
+
if evt.Action != "delete" {
96
+
// TODO: validate lexicon using catalog
97
+
var profile socialscrumble.BetaAccountProfile
98
+
if err := json.Unmarshal(evt.Record, &profile); err != nil {
99
+
logger.Warn("invalid profile record", "err", err)
100
+
return nil
101
+
}
102
+
}
103
+
case "social.scrumble.beta.collection.item":
104
+
_, err := syntax.ParseTID(evt.Rkey)
105
+
if err != nil {
106
+
return nil // skip
107
+
}
108
+
if evt.Action == "delete" {
109
+
if err := idx.store.DeleteItem(did, rkey); err != nil {
110
+
return err
111
+
}
112
+
} else {
113
+
// TODO: validate lexicon using catalog
114
+
var item socialscrumble.BetaCollectionItem
115
+
if err := json.Unmarshal(evt.Record, &item); err != nil {
116
+
logger.Warn("invalid item record", "err", err)
117
+
return nil
118
+
}
119
+
scene, err := syntax.ParseDID(item.Scene)
120
+
if err != nil {
121
+
logger.Warn("invalid scene DID", "err", err, "scene", item.Scene)
122
+
return nil
123
+
}
124
+
if !idx.WantScene(scene) {
125
+
return nil // skip
126
+
}
127
+
if err := idx.store.UpsertItem(scene, did, rkey, evt.CID); err != nil {
128
+
return err
129
+
}
130
+
}
131
+
default:
132
+
// ignore
133
+
return nil
134
+
}
135
+
136
+
// update record table for all collection types (if we got this far)
137
+
if evt.Action == "delete" {
138
+
if err := idx.store.DeleteRecord(did, collection, rkey); err != nil {
139
+
return err
140
+
}
141
+
} else {
142
+
// TODO: parse version CID?
143
+
if err := idx.store.UpsertRecord(did, collection, rkey, evt.CID, evt.Record); err != nil {
144
+
return err
145
+
}
146
+
}
147
+
return nil
148
+
}
149
+
150
+
// checks whether scene (DID) is in-scope for this server
151
+
func (idx *Indexer) WantScene(did syntax.DID) bool {
152
+
if len(idx.sceneDIDs) == 0 {
153
+
return true
154
+
}
155
+
for _, d := range idx.sceneDIDs {
156
+
if d == did {
157
+
return true
158
+
}
159
+
}
160
+
return false
161
+
}
162
+
163
+
func (idx *Indexer) Start(tapURL string) error {
164
+
ctx := context.Background()
165
+
166
+
slog.Info("starting indexer")
167
+
ws, err := tapclient.NewWebsocket(tapURL, idx.HandleMessage,
168
+
tapclient.WithLogger(slog.Default()),
169
+
tapclient.WithAcks(),
170
+
)
171
+
if err != nil {
172
+
return err
173
+
}
174
+
175
+
return ws.Run(ctx)
176
+
}
177
+
178
+
func (idx *Indexer) Shutdown() error {
179
+
slog.Info("shutting down indexer (UNIMPLEMENTED)")
180
+
return nil
181
+
}
+6
-1
main.go
+6
-1
main.go
···
54
54
&cli.StringFlag{
55
55
Name: "db-url",
56
56
Usage: "database connection string for relay database",
57
-
Value: "sqlite://data/relay/relay.sqlite",
57
+
Value: "sqlite://data/scrumble.sqlite",
58
58
Sources: cli.EnvVars("DATABASE_URL"),
59
59
},
60
60
&cli.IntFlag{
···
155
155
srvErr := make(chan error, 1)
156
156
go func() {
157
157
err := srv.StartHTTP(cmd.String("bind"))
158
+
srvErr <- err
159
+
}()
160
+
161
+
go func() {
162
+
err := srv.StartIndexer()
158
163
srvErr <- err
159
164
}()
160
165
+26
-9
server.go
+26
-9
server.go
···
7
7
"net/http"
8
8
"time"
9
9
10
+
"tangled.org/bnewbold.net/scrumble/indexer"
10
11
"tangled.org/bnewbold.net/scrumble/store"
11
12
12
13
"github.com/bluesky-social/indigo/atproto/identity"
14
+
"github.com/bluesky-social/indigo/atproto/syntax"
13
15
"github.com/bluesky-social/indigo/util/svcutil"
14
16
"github.com/labstack/echo/v4"
15
17
"github.com/labstack/echo/v4/middleware"
···
18
20
)
19
21
20
22
type Server struct {
21
-
logger *slog.Logger
22
-
dir identity.Directory
23
-
store *store.Store
23
+
logger *slog.Logger
24
+
dir identity.Directory
25
+
store *store.Store
26
+
indexer *indexer.Indexer
24
27
}
25
28
26
29
func NewServer(db *gorm.DB) (*Server, error) {
···
30
33
return nil, err
31
34
}
32
35
36
+
idx, err := indexer.NewIndexer(st, []syntax.DID{})
37
+
if err != nil {
38
+
return nil, err
39
+
}
40
+
33
41
return &Server{
34
-
logger: slog.Default(),
35
-
dir: identity.DefaultDirectory(),
36
-
store: st,
42
+
logger: slog.Default(),
43
+
dir: identity.DefaultDirectory(),
44
+
store: st,
45
+
indexer: idx,
37
46
}, nil
38
47
}
39
48
···
57
66
func (srv *Server) startWithListener(listen net.Listener) error {
58
67
e := echo.New()
59
68
e.HideBanner = true
69
+
e.HidePort = true
60
70
61
71
e.Use(middleware.CORSWithConfig(middleware.CORSConfig{
62
72
AllowOrigins: []string{"*"},
···
70
80
})
71
81
e.Use(middleware.LoggerWithConfig(middleware.DefaultLoggerConfig))
72
82
73
-
e.File("/robots.txt", "assets/robots.txt")
74
-
e.Static("/assets", "assets")
83
+
e.File("/robots.txt", "static/robots.txt")
84
+
e.Static("/static", "static")
75
85
76
86
e.Use(svcutil.MetricsMiddleware)
77
87
···
87
97
return e.StartServer(httpServer)
88
98
}
89
99
100
+
func (srv *Server) StartIndexer() error {
101
+
// TODO: wire through config
102
+
return srv.indexer.Start("ws://localhost:2480/channel")
103
+
}
104
+
90
105
func (srv *Server) Shutdown() []error {
91
106
var errs []error
92
107
93
-
// TODO: stop consumer
108
+
if err := srv.indexer.Shutdown(); err != nil {
109
+
errs = append(errs, err)
110
+
}
94
111
// TODO: stop echo
95
112
return errs
96
113
}
+1
static/robots.txt
+1
static/robots.txt
···
1
+
# hello friends!
+1
-1
store/models.go
+1
-1
store/models.go
+31
store/store.go
+31
store/store.go
···
1
1
package store
2
2
3
3
import (
4
+
"encoding/json"
5
+
4
6
"gorm.io/gorm"
7
+
8
+
"github.com/bluesky-social/indigo/atproto/syntax"
5
9
)
6
10
7
11
type Store struct {
···
44
48
}
45
49
return nil
46
50
}
51
+
52
+
// this method assumes the did, handle, and status have already been parsed/cleaned
53
+
func (st *Store) UpsertIdentity(did syntax.DID, handle syntax.Handle, upstreamStatus string) error {
54
+
var row Account
55
+
return st.db.Where(Account{DID: did.String()}).Assign(Account{Handle: handle.String(), UpstreamStatus: upstreamStatus}).FirstOrCreate(&row).Error
56
+
}
57
+
58
+
func (st *Store) UpsertRecord(did syntax.DID, collection syntax.NSID, rkey syntax.RecordKey, version string, val json.RawMessage) error {
59
+
var row Record
60
+
return st.db.Where(Record{AccountDID: did.String(), Collection: collection.String(), RKey: rkey.String()}).Assign(Record{Version: version, DataJSON: string(val)}).FirstOrCreate(&row).Error
61
+
}
62
+
63
+
func (st *Store) DeleteRecord(did syntax.DID, collection syntax.NSID, rkey syntax.RecordKey) error {
64
+
// TODO: only log on "record not found"?
65
+
return st.db.Delete(&Record{AccountDID: did.String(), Collection: collection.String(), RKey: rkey.String()}).Error
66
+
}
67
+
68
+
func (st *Store) UpsertItem(sceneDID syntax.DID, accountDID syntax.DID, rkey syntax.RecordKey, version string) error {
69
+
var row Item
70
+
return st.db.Where(Item{SceneDID: sceneDID.String(), AccountDID: accountDID.String(), RKey: rkey.String()}).Assign(Item{Version: version}).FirstOrCreate(&row).Error
71
+
}
72
+
73
+
func (st *Store) DeleteItem(accountDID syntax.DID, rkey syntax.RecordKey) error {
74
+
// TODO: only log on "record not found"?
75
+
var row Item
76
+
return st.db.Where("account_did = ? AND rkey = ?", accountDID, rkey).Delete(&row).Error
77
+
}