drawbu.dev / tangled
Monorepo for Tangled
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

appview: use blob based strings

Signed-off-by: Seongmin Lee <git@boltless.me>

Seongmin Lee (May 10, 2026, 12:53 AM +0900) 71f8a309 1297a3e0

+510 -138
+18
appview/db/db.go
··· 1695 1695 return nil 1696 1696 }) 1697 1697 1698 + orm.RunMigration(conn, logger, "use-blobs-for-string-files", func(tx *sql.Tx) error { 1699 + _, err := tx.Exec(` 1700 + create table string_files ( 1701 + id integer primary key autoincrement, 1702 + at_uri text not null, 1703 + name text not null, 1704 + content_ref text not null, 1705 + content_size integer not null, 1706 + content_mimetype text not null, 1707 + gzip_realsize integer, 1708 + gzip_realmime text, 1709 + gzip_realcontent text, -- decoded content of gzipped text blobs 1710 + foreign key (at_uri) references strings(at_uri) on delete cascade 1711 + ); 1712 + `) 1713 + return err 1714 + }) 1715 + 1698 1716 return &DB{ 1699 1717 db, 1700 1718 logger,
+117 -41
appview/db/strings.go
··· 10 10 "time" 11 11 12 12 "github.com/bluesky-social/indigo/atproto/syntax" 13 + lexutil "github.com/bluesky-social/indigo/lex/util" 14 + "github.com/ipfs/go-cid" 15 + "tangled.org/core/api/tangled" 13 16 "tangled.org/core/appview/models" 14 17 "tangled.org/core/orm" 15 18 ) ··· 61 64 } 62 65 if num == 0 { 63 66 return nil 67 + } 68 + 69 + log.Println("inserting files", len(s.Files)) 70 + 71 + _, err = tx.Exec(`delete from string_files where at_uri = ?`, s.AtUri()) 72 + if err != nil { 73 + return fmt.Errorf("deleting old files: %w", err) 74 + } 75 + 76 + vals := make([]string, len(s.Files)) 77 + args := make([]any, 0, len(s.Files)*8) 78 + for i, file := range s.Files { 79 + vals[i] = "(?, ?, ?, ?, ?, ?, ?, ?)" 80 + var gzipRealSize *int64 81 + var gzipRealMime *string 82 + var gzipRealContent *string 83 + if file.Gzip != nil { 84 + gzipRealSize = &file.Gzip.RealSize 85 + gzipRealMime = &file.Gzip.RealMime 86 + gzipRealContent = &file.Gzip.Content 87 + } 88 + args = append(args, 89 + s.AtUri(), 90 + file.Name, 91 + file.Content.Ref.String(), 92 + file.Content.Size, 93 + file.Content.MimeType, 94 + gzipRealSize, 95 + gzipRealMime, 96 + gzipRealContent, 97 + ) 98 + } 99 + _, err = tx.Exec( 100 + fmt.Sprintf( 101 + `insert into string_files ( 102 + at_uri, 103 + name, 104 + content_ref, 105 + content_size, 106 + content_mimetype, 107 + gzip_realsize, 108 + gzip_realmime, 109 + gzip_realcontent 110 + ) 111 + values %s`, 112 + strings.Join(vals, ","), 113 + ), 114 + args..., 115 + ) 116 + if err != nil { 117 + return fmt.Errorf("inserting files: %w", err) 64 118 } 65 119 66 120 if err := tx.Commit(); err != nil { ··· 195 249 i++ 196 250 } 197 251 198 - // // get files 199 - // { 200 - // rows, err := e.Query( 201 - // fmt.Sprintf( 202 - // `select at_uri, name, blob from string_files where at_uri in (%s) order by at_uri, id`, 203 - // inClause, 204 - // ), 205 - // args..., 206 - // ) 207 - // if err != nil { 208 - // return nil, fmt.Errorf("failed to execute string_files query: %w", err) 209 - // } 210 - // defer rows.Close() 211 - // 212 - // for rows.Next() { 213 - // var stringAt syntax.ATURI 214 - // var file models.String_File 215 - // file.Blob = &util.LexBlob{} 216 - // var gzipMimeType sql.Null[string] 217 - // var gzipSize sql.Null[int64] 218 - // if err := rows.Scan( 219 - // &stringAt, 220 - // &file.Name, 221 - // &blob, 222 - // ); err != nil { 223 - // return nil, fmt.Errorf("failed to execute string_files query: %w", err) 224 - // } 225 - // if gzipMimeType.Valid && gzipSize.Valid { 226 - // file.Gzip = &models.GzipInfo{ 227 - // MimeType: gzipMimeType.V, 228 - // Size: gzipSize.V, 229 - // } 230 - // } 231 - // if s, ok := stringMap[stringAt]; ok { 232 - // s.Files = append(s.Files, file) 233 - // } 234 - // } 235 - // if err = rows.Err(); err != nil { 236 - // return nil, fmt.Errorf("failed to execute string_files query: %w", err) 237 - // } 238 - // } 252 + // get files 253 + { 254 + rows, err := e.Query( 255 + fmt.Sprintf( 256 + `select 257 + at_uri, 258 + name, 259 + content_ref, 260 + content_size, 261 + content_mimetype, 262 + gzip_realsize, 263 + gzip_realmime, 264 + gzip_realcontent 265 + from string_files 266 + where at_uri in (%s) order by at_uri, id`, 267 + inClause, 268 + ), 269 + args..., 270 + ) 271 + if err != nil { 272 + return nil, fmt.Errorf("failed to execute string_files query: %w", err) 273 + } 274 + defer rows.Close() 275 + 276 + for rows.Next() { 277 + var stringAt syntax.ATURI 278 + var file models.String_File 279 + 280 + var contentRef string 281 + var gzipRealSize sql.Null[int64] 282 + var gzipRealMime, gzipRealContent sql.Null[string] 283 + if err := rows.Scan( 284 + &stringAt, 285 + &file.Name, 286 + &contentRef, 287 + &file.Content.Size, 288 + &file.Content.MimeType, 289 + &gzipRealSize, 290 + &gzipRealMime, 291 + &gzipRealContent, 292 + ); err != nil { 293 + return nil, fmt.Errorf("failed to execute string_files query: %w", err) 294 + } 295 + 296 + file.Content.Ref = lexutil.LexLink(cid.MustParse(contentRef)) 297 + 298 + if gzipRealMime.Valid && gzipRealSize.Valid && gzipRealContent.Valid { 299 + file.Gzip = &models.String_GzipInfo{ 300 + String_File_Gzip: tangled.String_File_Gzip{ 301 + RealMime: gzipRealMime.V, 302 + RealSize: gzipRealSize.V, 303 + }, 304 + Content: gzipRealContent.V, 305 + } 306 + } 307 + if s, ok := stringMap[stringAt]; ok { 308 + s.Files = append(s.Files, file) 309 + } 310 + } 311 + if err = rows.Err(); err != nil { 312 + return nil, fmt.Errorf("failed to execute string_files query: %w", err) 313 + } 314 + } 239 315 240 316 // get star counts 241 317 {
+38 -5
appview/ingester.go
··· 1 1 package appview 2 2 3 3 import ( 4 + "compress/gzip" 4 5 "context" 5 6 "database/sql" 6 7 "encoding/json" ··· 31 32 "tangled.org/core/appview/notify" 32 33 "tangled.org/core/appview/serververify" 33 34 "tangled.org/core/appview/validator" 35 + "tangled.org/core/blobstore" 34 36 "tangled.org/core/idresolver" 35 37 "tangled.org/core/orm" 36 38 "tangled.org/core/rbac" ··· 40 42 Db db.DbWrapper 41 43 Enforcer *rbac.Enforcer 42 44 IdResolver *idresolver.Resolver 45 + BlobStore blobstore.BlobStore 43 46 Cache *cache.Cache 44 47 Config *config.Config 45 48 Logger *slog.Logger ··· 90 93 case tangled.KnotNSID: 91 94 err = i.ingestKnot(e) 92 95 case tangled.StringNSID: 93 - err = i.ingestString(e) 96 + err = i.ingestString(ctx, e) 94 97 case tangled.RepoIssueNSID: 95 98 err = i.ingestIssue(ctx, e) 96 99 case tangled.RepoPullNSID: ··· 797 800 return nil 798 801 } 799 802 800 - func (i *Ingester) ingestString(e *jmodels.Event) error { 803 + func (i *Ingester) ingestString(ctx context.Context, e *jmodels.Event) error { 801 804 did := e.Did 802 805 rkey := e.Commit.RKey 803 806 ··· 821 824 return err 822 825 } 823 826 824 - string, err := models.StringFromRecord(syntax.DID(did), syntax.RecordKey(rkey), syntax.CID(e.Commit.CID), record) 827 + str, err := models.StringFromRecord(syntax.DID(did), syntax.RecordKey(rkey), syntax.CID(e.Commit.CID), record) 825 828 if err != nil { 826 829 return fmt.Errorf("failed to parse string record: %w", err) 827 830 } 828 - if err = string.Validate(); err != nil { 831 + if err = str.Validate(); err != nil { 829 832 l.Error("invalid record", "err", err) 830 833 return err 831 834 } 832 835 833 - if err = db.AddString(ddb, string); err != nil { 836 + g, gctx := errgroup.WithContext(ctx) 837 + for idx, file := range str.Files { 838 + if file.Gzip == nil { 839 + continue 840 + } 841 + g.Go(func() error { 842 + blob, err := i.BlobStore.GetBlob(gctx, str.Did, cid.Cid(file.Content.Ref)) 843 + if err != nil { 844 + return fmt.Errorf("files[%d]: failed to fetch blob: %w", idx, err) 845 + } 846 + defer blob.Close() 847 + gzr, err := gzip.NewReader(blob) 848 + if err != nil { 849 + return fmt.Errorf("files[%d]: invalid gzip stream: %w", idx, err) 850 + } 851 + gzr.Close() 852 + 853 + content, err := io.ReadAll(gzr) 854 + if err != nil { 855 + return fmt.Errorf("files[%d]: failed to read blob: %w", idx, err) 856 + } 857 + file.Gzip.Content = string(content) 858 + str.Files[idx] = file 859 + return nil 860 + }) 861 + } 862 + if err := g.Wait(); err != nil { 863 + return err 864 + } 865 + 866 + if err = db.AddString(ddb, str); err != nil { 834 867 l.Error("failed to add string", "err", err) 835 868 return err 836 869 }
+51 -19
appview/models/string.go
··· 37 37 Stats *StringStats 38 38 } 39 39 40 - // TODO: replace this with [tangled.String_File] 40 + // String_File is [tangled.String_File] with optional prefetched & decompressed text blob content 41 41 type String_File struct { 42 - Name string 43 - Blob *lexutil.LexBlob 44 - Gzip *GzipInfo 42 + Name string 43 + Content lexutil.LexBlob 44 + Gzip *String_GzipInfo 45 45 } 46 - type GzipInfo struct { 47 - MimeType string 48 - Size int64 46 + type String_GzipInfo struct { 47 + tangled.String_File_Gzip 48 + // Optional uncompressed content. 49 + // Populated when the content is first requested. 50 + Content string 49 51 } 50 52 51 53 func (s *String) AtUri() syntax.ATURI { ··· 53 55 } 54 56 55 57 func (s *String) AsRecord() *tangled.String { 56 - var description string 57 - if s.Description != nil { 58 - description = *s.Description 58 + var files []*tangled.String_File 59 + for _, f := range s.Files { 60 + var gzip *tangled.String_File_Gzip 61 + if f.Gzip != nil { 62 + gzip = &tangled.String_File_Gzip{ 63 + RealSize: f.Gzip.RealSize, 64 + RealMime: f.Gzip.RealMime, 65 + } 66 + } 67 + files = append(files, &tangled.String_File{ 68 + Name: f.Name, 69 + Content: &f.Content, 70 + Gzip: gzip, 71 + }) 59 72 } 60 73 return &tangled.String{ 61 - Filename: s.FileName, 62 - Description: description, 63 - Contents: s.FileContent, 74 + Title: s.Title, 75 + Description: s.Description, 76 + Files: files, 64 77 CreatedAt: s.Created.Format(time.RFC3339), 65 78 } 66 79 } ··· 102 115 return len(s.Files) == 0 103 116 } 104 117 118 + // StringFromRecord creates [String] from [tangled.String]. 119 + // NOTE: This won't prefetch blobs 105 120 func StringFromRecord(did syntax.DID, rkey syntax.RecordKey, cid syntax.CID, record tangled.String) (String, error) { 106 121 created, err := time.Parse(time.RFC3339, record.CreatedAt) 107 122 if err != nil { 108 123 return String{}, fmt.Errorf("invalid createdAt: %w", err) 109 124 } 110 - var description *string 111 - if record.Description != "" { 112 - description = &record.Description 125 + var files []String_File 126 + for _, f := range record.Files { 127 + var gzip *String_GzipInfo 128 + if f.Gzip != nil { 129 + gzip = &String_GzipInfo{String_File_Gzip: *f.Gzip} 130 + } 131 + files = append(files, String_File{ 132 + Name: f.Name, 133 + Content: *f.Content, 134 + Gzip: gzip, 135 + }) 113 136 } 114 137 return String{ 115 138 Did: did, 116 139 Rkey: rkey, 117 140 Cid: &cid, 118 - Description: description, 141 + Title: record.Title, 142 + Description: record.Description, 143 + Files: files, 119 144 Created: created, 120 - FileName: record.Filename, 121 - FileContent: record.Contents, 145 + FileName: stringPtr(record.Filename), 146 + FileContent: stringPtr(record.Contents), 122 147 }, nil 123 148 } 124 149 ··· 131 156 LineCount int 132 157 ByteCount int 133 158 } 159 + 160 + func stringPtr(s *string) string { 161 + if s == nil { 162 + return "" 163 + } 164 + return *s 165 + }
+7 -6
appview/state/router.go
··· 313 313 logger := log.SubLogger(s.logger, "strings") 314 314 315 315 strs := &avstrings.Strings{ 316 - Db: s.db, 317 - OAuth: s.oauth, 318 - Pages: s.pages, 319 - Dir: s.idResolver.Directory(), 320 - Notifier: s.notifier, 321 - Logger: logger, 316 + Db: s.db, 317 + OAuth: s.oauth, 318 + Pages: s.pages, 319 + Dir: s.idResolver.Directory(), 320 + Notifier: s.notifier, 321 + Logger: logger, 322 + BlobStore: s.blobStore, 322 323 } 323 324 324 325 return strs.Router(mw)
+6
appview/state/state.go
··· 31 31 "tangled.org/core/appview/reporesolver" 32 32 "tangled.org/core/appview/validator" 33 33 xrpcclient "tangled.org/core/appview/xrpcclient" 34 + "tangled.org/core/blobstore" 34 35 "tangled.org/core/consts" 35 36 "tangled.org/core/eventconsumer" 36 37 "tangled.org/core/idresolver" ··· 59 60 enforcer *rbac.Enforcer 60 61 pages *pages.Pages 61 62 idResolver *idresolver.Resolver 63 + blobStore blobstore.BlobStore 62 64 rdb *cache.Cache 63 65 mentionsResolver *mentions.Resolver 64 66 posthog posthog.Client ··· 118 120 119 121 mentionsResolver := mentions.New(config, res, d, log.SubLogger(logger, "mentionsResolver")) 120 122 123 + blobStore := blobstore.NewPdsBlobStore(res.Directory()) 124 + 121 125 wrapper := db.DbWrapper{Execer: d} 122 126 jc, err := jetstream.NewJetstreamClient( 123 127 config.Jetstream.Endpoint, ··· 179 183 Db: wrapper, 180 184 Enforcer: enforcer, 181 185 IdResolver: res, 186 + BlobStore: blobStore, 182 187 Cache: rdb, 183 188 Config: config, 184 189 Logger: log.SubLogger(logger, "ingester"), ··· 220 225 enforcer: enforcer, 221 226 pages: pages, 222 227 idResolver: res, 228 + blobStore: blobStore, 223 229 rdb: rdb, 224 230 mentionsResolver: mentionsResolver, 225 231 posthog: posthog,
+180 -67
appview/strings/strings.go
··· 2 2 3 3 import ( 4 4 "bytes" 5 + "compress/gzip" 5 6 "context" 6 7 "database/sql" 7 8 "errors" ··· 21 22 "tangled.org/core/appview/oauth" 22 23 "tangled.org/core/appview/pages" 23 24 "tangled.org/core/appview/pages/markup" 25 + "tangled.org/core/blobstore" 24 26 "tangled.org/core/orm" 25 27 "tangled.org/core/tid" 28 + "tangled.org/core/xrpc" 26 29 27 30 "github.com/bluesky-social/indigo/api/agnostic" 28 31 "github.com/bluesky-social/indigo/api/atproto" 29 32 "github.com/bluesky-social/indigo/atproto/identity" 30 33 "github.com/bluesky-social/indigo/atproto/syntax" 31 - "github.com/bluesky-social/indigo/xrpc" 34 + indigoxrpc "github.com/bluesky-social/indigo/xrpc" 32 35 "github.com/go-chi/chi/v5" 36 + "github.com/ipfs/go-cid" 33 37 34 38 comatproto "github.com/bluesky-social/indigo/api/atproto" 35 39 lexutil "github.com/bluesky-social/indigo/lex/util" 36 40 ) 41 + 42 + const ApplicationGzip = "application/gzip" 37 43 38 44 type Strings struct { 39 - Db *db.DB 40 - OAuth *oauth.OAuth 41 - Pages *pages.Pages 42 - Dir identity.Directory 43 - Logger *slog.Logger 44 - Notifier notify.Notifier 45 + Db *db.DB 46 + OAuth *oauth.OAuth 47 + Pages *pages.Pages 48 + Dir identity.Directory 49 + BlobStore blobstore.BlobStore 50 + Logger *slog.Logger 51 + Notifier notify.Notifier 45 52 } 46 53 47 54 func (s *Strings) Router(mw *middleware.Middleware) http.Handler { ··· 117 124 return 118 125 } 119 126 120 - string, err := db.GetString(s.Db, orm.FilterEq("did", id.DID), orm.FilterEq("rkey", rkey)) 127 + str, err := db.GetString(s.Db, orm.FilterEq("did", id.DID), orm.FilterEq("rkey", rkey)) 121 128 if errors.Is(err, sql.ErrNoRows) { 122 129 s.Pages.Error404(w) 123 130 return ··· 127 134 return 128 135 } 129 136 130 - ctx := context.WithValue(r.Context(), stringCtxKey{}, string) 137 + ctx := context.WithValue(r.Context(), stringCtxKey{}, str) 131 138 next.ServeHTTP(w, r.WithContext(ctx)) 132 139 }) 133 140 } ··· 170 177 l := s.Logger.With("handler", "SingleString") 171 178 ctx := r.Context() 172 179 173 - string, ok := stringFromContext(ctx) 180 + str, ok := stringFromContext(ctx) 174 181 if !ok { 175 182 l.Error("malformed middleware. string missing") 176 183 s.Pages.Error404(w) 177 184 return 178 185 } 179 186 180 - starCount, err := db.GetStarCount(s.Db, string.AtUri()) 187 + starCount, err := db.GetStarCount(s.Db, str.AtUri()) 181 188 if err != nil { 182 189 l.Error("failed to get star count", "err", err) 183 190 } 184 191 user := s.OAuth.GetMultiAccountUser(r) 185 192 isStarred := false 186 193 if user != nil { 187 - isStarred = db.GetStarStatus(s.Db, user.Did, string.AtUri()) 194 + isStarred = db.GetStarStatus(s.Db, user.Did, str.AtUri()) 188 195 } 189 196 190 - comments, err := db.GetComments(s.Db, orm.FilterEq("subject_uri", string.AtUri())) 197 + comments, err := db.GetComments(s.Db, orm.FilterEq("subject_uri", str.AtUri())) 191 198 if err != nil { 192 199 l.Error("failed to get comments", "err", err) 193 200 } ··· 223 230 224 231 var files []pages.StringFileFragmentParams 225 232 226 - if string.IsLegacySingleFile() { 233 + if str.IsLegacySingleFile() { 227 234 files = []pages.StringFileFragmentParams{ 228 - s.makeFileFragmentParams(&string, string.FileName, string.FileContent, false), 235 + s.makeFileFragmentParams(&str, str.FileName, str.FileContent, false), 229 236 } 230 237 } else { 231 - files = make([]pages.StringFileFragmentParams, len(string.Files)) 232 - for i, file := range string.Files { 233 - // TODO: read blob 234 - content := "" 235 - files[i] = s.makeFileFragmentParams(&string, file.Name, content, false) 238 + files = make([]pages.StringFileFragmentParams, len(str.Files)) 239 + for i, file := range str.Files { 240 + var content string 241 + if file.Gzip != nil { 242 + content = file.Gzip.Content 243 + } else { 244 + blob, err := s.BlobStore.GetBlob(r.Context(), str.Did, cid.Cid(file.Content.Ref)) 245 + if err != nil { 246 + l.Warn("failed to fetch blob", "err", err) 247 + http.NotFound(w, r) 248 + return 249 + } 250 + defer blob.Close() 251 + 252 + contentBytes, err := io.ReadAll(blob) 253 + if err != nil { 254 + l.Error("failed to read blob", "err", err) 255 + } 256 + content = string(contentBytes) 257 + } 258 + 259 + files[i] = s.makeFileFragmentParams(&str, file.Name, content, false) 236 260 } 237 261 } 238 262 239 263 err = s.Pages.SingleString(w, pages.SingleStringParams{ 240 264 LoggedInUser: user, 241 - String: &string, 265 + String: &str, 242 266 FileParams: files, 243 267 IsStarred: isStarred, 244 268 StarCount: starCount, ··· 288 312 } else { 289 313 files = make([]pages.StringFileEditFragmentParams, len(oldString.Files)) 290 314 for i, file := range oldString.Files { 291 - // TODO: read blob 292 - content := "" 315 + var content string 316 + if file.Gzip != nil { 317 + content = file.Gzip.Content 318 + } else { 319 + blob, err := s.BlobStore.GetBlob(r.Context(), oldString.Did, cid.Cid(file.Content.Ref)) 320 + if err != nil { 321 + l.Warn("failed to fetch blob", "err", err) 322 + http.NotFound(w, r) 323 + return 324 + } 325 + defer blob.Close() 326 + 327 + contentBytes, err := io.ReadAll(blob) 328 + if err != nil { 329 + l.Error("failed to read blob", "err", err) 330 + } 331 + content = string(contentBytes) 332 + } 293 333 files[i] = pages.StringFileEditFragmentParams{ 294 334 Name: file.Name, 295 335 Content: content, 296 - Size: uint64(file.Blob.Size), 336 + Size: uint64(file.Content.Size), 297 337 } 298 338 } 299 339 } ··· 333 373 return 334 374 } 335 375 336 - newString := oldString 337 - newString.Title = title 338 - newString.Description = description 339 - newString.FileName = filename 340 - newString.FileContent = content 341 - 342 376 client, err := s.OAuth.AuthorizedClient(r) 343 377 if err != nil { 344 378 fail("Failed to create record.", err) 345 379 return 346 380 } 347 381 382 + blob, err := xrpc.RepoUploadBlob(ctx, client, gz(content), ApplicationGzip) 383 + if err != nil { 384 + fail("Failed to create record.", err) 385 + return 386 + } 387 + 388 + newString := oldString 389 + newString.Title = title 390 + newString.Description = description 391 + newString.Files = []models.String_File{ 392 + { 393 + Name: filename, 394 + Content: *blob.Blob, 395 + Gzip: &models.String_GzipInfo{ 396 + String_File_Gzip: tangled.String_File_Gzip{ 397 + RealMime: "text/plain", 398 + RealSize: int64(len(content)), 399 + }, 400 + Content: content, 401 + }, 402 + }, 403 + } 404 + 348 405 // first replace the existing record in the PDS 349 406 var exCid string 350 407 if newString.Cid != nil { ··· 390 447 391 448 func (s *Strings) create(w http.ResponseWriter, r *http.Request) { 392 449 l := s.Logger.With("handler", "create") 450 + ctx := r.Context() 393 451 user := s.OAuth.GetMultiAccountUser(r) 394 452 395 453 switch r.Method { ··· 428 486 return 429 487 } 430 488 431 - string := models.String{ 432 - Did: syntax.DID(user.Did), 433 - Rkey: syntax.RecordKey(tid.TID()), 434 - Title: title, 435 - Description: description, 436 - FileName: filename, 437 - FileContent: content, 438 - Created: time.Now(), 489 + client, err := s.OAuth.AuthorizedClient(r) 490 + if err != nil { 491 + fail("Failed to create record.", err) 492 + return 439 493 } 440 494 441 - client, err := s.OAuth.AuthorizedClient(r) 495 + blob, err := xrpc.RepoUploadBlob(ctx, client, gz(content), ApplicationGzip) 442 496 if err != nil { 443 497 fail("Failed to create record.", err) 444 498 return 445 499 } 446 500 447 - resp, err := comatproto.RepoPutRecord(r.Context(), client, &atproto.RepoPutRecord_Input{ 501 + newString := models.String{ 502 + Did: syntax.DID(user.Did), 503 + Rkey: syntax.RecordKey(tid.TID()), 504 + Title: title, 505 + Description: description, 506 + Files: []models.String_File{ 507 + { 508 + Name: filename, 509 + Content: *blob.Blob, 510 + Gzip: &models.String_GzipInfo{ 511 + String_File_Gzip: tangled.String_File_Gzip{ 512 + RealMime: "text/plain", 513 + RealSize: int64(len(content)), 514 + }, 515 + Content: content, 516 + }, 517 + }, 518 + }, 519 + Created: time.Now(), 520 + } 521 + 522 + resp, err := comatproto.RepoPutRecord(ctx, client, &atproto.RepoPutRecord_Input{ 448 523 Collection: tangled.StringNSID, 449 - Repo: string.Did.String(), 450 - Rkey: string.Rkey.String(), 451 - Record: &lexutil.LexiconTypeDecoder{Val: string.AsRecord()}, 524 + Repo: newString.Did.String(), 525 + Rkey: newString.Rkey.String(), 526 + Record: &lexutil.LexiconTypeDecoder{Val: newString.AsRecord()}, 452 527 }) 453 528 if err != nil { 454 529 fail("Failed to create record.", err) 455 530 return 456 531 } 457 532 l := l.With("aturi", resp.Uri) 458 - l.Info("created record") 533 + l.Info("created record", "files", len(newString.Files)) 459 534 460 535 // insert into DB 461 - if err = db.AddString(s.Db, string); err != nil { 536 + if err = db.AddString(s.Db, newString); err != nil { 462 537 fail("Failed to create string.", err) 463 538 return 464 539 } 465 540 466 - s.Notifier.NewString(r.Context(), &string) 541 + s.Notifier.NewString(ctx, &newString) 467 542 468 543 // successful 469 - s.Pages.HxRedirect(w, fmt.Sprintf("/strings/%s/%s", string.Did, string.Rkey)) 544 + s.Pages.HxRedirect(w, fmt.Sprintf("/strings/%s/%s", newString.Did, newString.Rkey)) 470 545 } 471 546 } 472 547 ··· 533 608 l := s.Logger.With("handler", "FileRaw") 534 609 ctx := r.Context() 535 610 536 - string, ok := stringFromContext(ctx) 611 + str, ok := stringFromContext(ctx) 537 612 if !ok { 538 613 l.Error("malformed middleware. string missing") 539 614 s.Pages.Error404(w) ··· 541 616 } 542 617 filename := chi.URLParam(r, "filename") 543 618 544 - if string.IsLegacySingleFile() { 545 - if filename != string.FileName { 619 + if str.IsLegacySingleFile() { 620 + if filename != str.FileName { 546 621 http.NotFound(w, r) 547 622 return 548 623 } 549 624 w.Header().Set("Content-Type", "text/plain; charset=utf-8") 550 625 w.Header().Set("Content-Disposition", fmt.Sprintf("inline; filename=%q", filename)) 551 - w.Header().Set("Content-Length", strconv.Itoa(len(string.FileContent))) 552 - _, err := w.Write([]byte(string.FileContent)) 626 + w.Header().Set("Content-Length", strconv.Itoa(len(str.FileContent))) 627 + _, err := w.Write([]byte(str.FileContent)) 553 628 if err != nil { 554 629 l.Error("failed to write raw response", "err", err) 555 630 } 556 631 } else { 557 - file, ok := string.FileByName(filename) 632 + file, ok := str.FileByName(filename) 558 633 if !ok { 559 634 http.NotFound(w, r) 560 635 return 561 636 } 562 637 563 - content := "" 638 + mimeType := file.Content.MimeType 639 + size := file.Content.Size 640 + 641 + var reader io.Reader 642 + if file.Gzip != nil { 643 + reader = strings.NewReader(file.Gzip.Content) 644 + } else { 645 + blob, err := s.BlobStore.GetBlob(r.Context(), str.Did, cid.Cid(file.Content.Ref)) 646 + if err != nil { 647 + l.Warn("failed to fetch blob", "err", err) 648 + http.NotFound(w, r) 649 + return 650 + } 651 + defer blob.Close() 652 + reader = blob 653 + } 564 654 565 - w.Header().Set("Content-Type", "text/plain; charset=utf-8") 655 + w.Header().Set("Content-Type", mimeType) 566 656 w.Header().Set("Content-Disposition", fmt.Sprintf("inline; filename=%q", filename)) 567 - w.Header().Set("Content-Length", strconv.FormatInt(file.Blob.Size, 10)) 568 - _, err := w.Write([]byte(content)) 569 - if err != nil { 657 + w.Header().Set("Content-Length", strconv.FormatInt(size, 10)) 658 + if _, err := io.Copy(w, reader); err != nil { 570 659 l.Error("failed to write raw response", "err", err) 571 660 } 572 661 } ··· 611 700 l := s.Logger.With("handler", "FileFragment") 612 701 ctx := r.Context() 613 702 614 - string, ok := stringFromContext(ctx) 703 + str, ok := stringFromContext(ctx) 615 704 if !ok { 616 705 l.Error("malformed middleware. string missing") 617 706 http.NotFound(w, r) ··· 621 710 forceCode := r.URL.Query().Get("code") == "true" 622 711 623 712 var params pages.StringFileFragmentParams 624 - if string.IsLegacySingleFile() { 625 - if filename != string.FileName { 713 + if str.IsLegacySingleFile() { 714 + if filename != str.FileName { 626 715 http.NotFound(w, r) 627 716 return 628 717 } 629 - params = s.makeFileFragmentParams(&string, string.FileName, string.FileContent, forceCode) 718 + params = s.makeFileFragmentParams(&str, str.FileName, str.FileContent, forceCode) 630 719 } else { 631 - file, ok := string.FileByName(filename) 720 + file, ok := str.FileByName(filename) 632 721 if !ok { 633 722 l.Error("malformed middleware. string missing") 634 723 http.NotFound(w, r) 635 724 return 636 725 } 637 726 638 - // TODO: read blob 639 - content := "" 727 + var content string 728 + if file.Gzip != nil && file.Gzip.Content != "" { 729 + content = file.Gzip.Content 730 + } else { 731 + blob, err := s.BlobStore.GetBlob(r.Context(), str.Did, cid.Cid(file.Content.Ref)) 732 + if err != nil { 733 + l.Warn("failed to fetch blob", "err", err) 734 + http.NotFound(w, r) 735 + return 736 + } 737 + defer blob.Close() 640 738 641 - params = s.makeFileFragmentParams(&string, file.Name, content, forceCode) 739 + contentBytes, err := io.ReadAll(blob) 740 + if err != nil { 741 + l.Error("failed to read blob", "err", err) 742 + } 743 + content = string(contentBytes) 744 + } 745 + 746 + params = s.makeFileFragmentParams(&str, file.Name, content, forceCode) 642 747 } 643 748 s.Pages.StringFileFragment(w, params) 644 749 } ··· 653 758 return "", err 654 759 } 655 760 656 - xrpcc := xrpc.Client{Host: ident.PDSEndpoint()} 761 + xrpcc := indigoxrpc.Client{Host: ident.PDSEndpoint()} 657 762 out, err := agnostic.RepoGetRecord(ctx, &xrpcc, "", uri.Collection().String(), ident.DID.String(), uri.RecordKey().String()) 658 763 if err != nil { 659 764 return "", err ··· 669 774 670 775 return cid, nil 671 776 } 777 + 778 + func gz(s string) io.Reader { 779 + var b bytes.Buffer 780 + w := gzip.NewWriter(&b) 781 + w.Write([]byte(s)) 782 + w.Close() 783 + return &b 784 + }
+13
blobstore/blobstore.go
··· 1 + package blobstore 2 + 3 + import ( 4 + "context" 5 + "io" 6 + 7 + "github.com/bluesky-social/indigo/atproto/syntax" 8 + "github.com/ipfs/go-cid" 9 + ) 10 + 11 + type BlobStore interface { 12 + GetBlob(ctx context.Context, did syntax.DID, cid cid.Cid) (io.ReadCloser, error) 13 + }
+46
blobstore/pds.go
··· 1 + package blobstore 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "io" 7 + "net/http" 8 + "net/url" 9 + 10 + "github.com/bluesky-social/indigo/atproto/identity" 11 + "github.com/bluesky-social/indigo/atproto/syntax" 12 + "github.com/ipfs/go-cid" 13 + ) 14 + 15 + type Pds struct { 16 + dir identity.Directory 17 + } 18 + 19 + func NewPdsBlobStore(dir identity.Directory) *Pds { 20 + return &Pds{dir} 21 + } 22 + 23 + func (s *Pds) GetBlob(ctx context.Context, did syntax.DID, cid cid.Cid) (io.ReadCloser, error) { 24 + id, err := s.dir.LookupDID(ctx, did) 25 + if err != nil { 26 + return nil, err 27 + } 28 + 29 + url, _ := url.Parse(fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob", id.PDSEndpoint())) 30 + q := url.Query() 31 + q.Set("did", did.String()) 32 + q.Set("cid", cid.String()) 33 + url.RawQuery = q.Encode() 34 + 35 + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url.String(), nil) 36 + resp, err := http.DefaultClient.Do(req) 37 + if err != nil { 38 + return nil, err 39 + } 40 + 41 + if resp.StatusCode != http.StatusOK { 42 + return nil, fmt.Errorf("unexpected status: %s", resp.Status) 43 + } 44 + 45 + return resp.Body, nil 46 + }
+34
blobstore/porxie.go
··· 1 + package blobstore 2 + 3 + import ( 4 + "context" 5 + "fmt" 6 + "io" 7 + "net/http" 8 + "path" 9 + 10 + "github.com/bluesky-social/indigo/atproto/syntax" 11 + "github.com/ipfs/go-cid" 12 + ) 13 + 14 + type Porxie struct { 15 + url string 16 + } 17 + 18 + func NewPorxieBlobStore(url string) *Porxie { 19 + return &Porxie{url} 20 + } 21 + 22 + func (s *Porxie) GetBlob(ctx context.Context, did syntax.DID, cid cid.Cid) (io.ReadCloser, error) { 23 + url := path.Join(s.url, did.String(), cid.String()) 24 + resp, err := http.Get(url) 25 + if err != nil { 26 + return nil, err 27 + } 28 + 29 + if resp.StatusCode != http.StatusOK { 30 + return nil, fmt.Errorf("unexpected status: %s", resp.Status) 31 + } 32 + 33 + return resp.Body, nil 34 + }