commit c9b3ba7a061dea6a8c285aed0a44060d6e256c8f · hailey.at/photocopy

+44 -10
backfiller.go
···

       7
       7
        
       	"fmt"

     

       8
       8
        
       	"io"

     

       9
       9
        
       	"net/http"

     

       10
       10
       +
       	"runtime"

     

       10
       11
        
       	"strings"

     

       11
       12
        
       	"sync"

     

       12
       13
        
       	"time"

     
···

       20
       21
        
       	"go.uber.org/ratelimit"

     

       21
       22
        
       )

     

       22
       23
        
       

     

       24
       24
       +
       type ProcessJob struct {

     

       25
       25
       +
       	did       string

     

       26
       26
       +
       	repoBytes []byte

     

       27
       27
       +
       }

     

       28
       28
       +
       

     

       23
       29
        
       type RepoDownloader struct {

     

       24
       24
       -
       	clients    map[string]*http.Client

     

       25
       25
       -
       	rateLimits map[string]ratelimit.Limiter

     

       26
       26
       -
       	mu         sync.RWMutex

     

       27
       27
       -
       	p          *Photocopy

     

       30
       30
       +
       	clients     map[string]*http.Client

     

       31
       31
       +
       	rateLimits  map[string]ratelimit.Limiter

     

       32
       32
       +
       	processChan chan ProcessJob

     

       33
       33
       +
       	mu          sync.RWMutex

     

       34
       34
       +
       	p           *Photocopy

     

       28
       35
        
       }

     

       29
       36
        
       

     

       30
       37
        
       func NewRepoDownloader(p *Photocopy) *RepoDownloader {

     

       31
       38
        
       	return &RepoDownloader{

     

       32
       32
       -
       		clients:    make(map[string]*http.Client),

     

       33
       33
       -
       		rateLimits: make(map[string]ratelimit.Limiter),

     

       34
       34
       -
       		p:          p,

     

       39
       39
       +
       		clients:     make(map[string]*http.Client),

     

       40
       40
       +
       		rateLimits:  make(map[string]ratelimit.Limiter),

     

       41
       41
       +
       		p:           p,

     

       42
       42
       +
       		processChan: make(chan ProcessJob, 1000),

     

       35
       43
        
       	}

     

       36
       44
        
       }

     

       37
       45
        
       

     
···

       122
       130
        
       	bs := atproto_repo.NewTinyBlockstore()

     

       123
       131
        
       	cs, err := car.NewCarReader(bytes.NewReader(b))

     

       124
       132
        
       	if err != nil {

     

       133
       133
       +
       		fmt.Println("error opening car", err)

     

       125
       134
        
       		return fmt.Errorf("error opening car: %v\n", err)

     

       126
       135
        
       	}

     

       127
       136
        
       

     
···

       134
       143
        
       

     

       135
       144
        
       	r, err := repo.OpenRepo(context.TODO(), bs, cs.Header.Roots[0])

     

       136
       145
        
       	if err != nil || r == nil {

     

       146
       146
       +
       		fmt.Println("error opening repo", err)

     

       137
       147
        
       		fmt.Printf("could not open repo: %v", err)

     

       138
       148
        
       		return nil

     

       139
       149
        
       	}

     
···

       238
       248
        
       	Status   string `json:"status"`

     

       239
       249
        
       }

     

       240
       250
        
       

     

       251
       251
       +
       func (p *Photocopy) runProcessRepoWorker(ctx context.Context, jobs <-chan ProcessJob) {

     

       252
       252
       +
       	for j := range jobs {

     

       253
       253
       +
       		p.processRepo(ctx, j.repoBytes, j.did)

     

       254
       254
       +
       	}

     

       255
       255
       +
       }

     

       256
       256
       +
       

     

       241
       257
        
       func (p *Photocopy) runBackfiller(ctx context.Context) error {

     

       242
       258
        
       	startTime := time.Now()

     

       243
       259
        
       

     

       244
       260
        
       	fmt.Println("querying clickhouse for dids and services...")

     

       245
       261
        
       

     

       262
       262
       +
       	var alreadyFetched []string

     

       263
       263
       +
       	if err := p.conn.Select(ctx, &alreadyFetched, "SELECT DISTINCT(did) FROM default.record WHERE created_at < '2025-07-01'"); err != nil {

     

       264
       264
       +
       		return err

     

       265
       265
       +
       	}

     

       266
       266
       +
       

     

       267
       267
       +
       	alreadyFetchedMap := map[string]bool{}

     

       268
       268
       +
       	for _, d := range alreadyFetched {

     

       269
       269
       +
       		alreadyFetchedMap[d] = true

     

       270
       270
       +
       	}

     

       271
       271
       +
       

     

       272
       272
       +
       	fmt.Println("getting dids")

     

       273
       273
       +
       

     

       246
       274
        
       	var hostsCursor string

     

       247
       275
        
       	var sevs []ListServicesResponseItem

     

       248
       276
        
       	for {

     
···

       299
       327
        
       	downloader := NewRepoDownloader(p)

     

       300
       328
        
       	serviceDids := map[string][]string{}

     

       301
       329
        
       

     

       330
       330
       +
       	for range runtime.NumCPU() / 2 {

     

       331
       331
       +
       		go p.runProcessRepoWorker(ctx, downloader.processChan)

     

       332
       332
       +
       	}

     

       333
       333
       +
       

     

       302
       334
        
       	wg := sync.WaitGroup{}

     

       303
       335
        
       	mplk := sync.Mutex{}

     

       304
       336
        
       	for s := range servicesDids {

     
···

       312
       344
        
       			}

     

       313
       345
        
       			dids := []string{}

     

       314
       346
        
       			for _, r := range repos {

     

       347
       347
       +
       				if alreadyFetchedMap[r.Did] {

     

       348
       348
       +
       					skipped++

     

       349
       349
       +
       					continue

     

       350
       350
       +
       				}

     

       315
       351
        
       				dids = append(dids, r.Did)

     

       316
       352
        
       			}

     

       317
       353
        
       			mplk.Lock()

     
···

       365
       401
        
       				}

     

       366
       402
        
       

     

       367
       403
        
       				go func(b []byte, did string) {

     

       368
       368
       -
       					if err := p.processRepo(ctx, b, did); err != nil {

     

       369
       369
       -
       						fmt.Printf("error processing backfill record: %v\n", err)

     

       370
       370
       -
       					}

     

       404
       404
       +
       					downloader.processChan <- ProcessJob{repoBytes: b, did: did}

     

       371
       405
        
       				}(b, did)

     

       372
       406
        
       

     

       373
       407
        
       				processed++