forked from
willdot.net/cocoon
A fork of the Cocoon PDS but being made more distributed.
1package server
2
3import (
4 "bytes"
5 "context"
6 "io"
7 "slices"
8 "strings"
9
10 "github.com/bluesky-social/indigo/atproto/syntax"
11 "github.com/haileyok/cocoon/internal/helpers"
12 "github.com/haileyok/cocoon/models"
13 blocks "github.com/ipfs/go-block-format"
14 "github.com/ipfs/go-cid"
15 "github.com/ipld/go-car"
16 "github.com/labstack/echo/v4"
17)
18
19func (s *Server) handleRepoImportRepo(e echo.Context) error {
20 ctx := e.Request().Context()
21 logger := s.logger.With("name", "handleImportRepo")
22
23 urepo := e.Get("repo").(*models.RepoActor)
24
25 b, err := io.ReadAll(e.Request().Body)
26 if err != nil {
27 logger.Error("could not read bytes in import request", "error", err)
28 return helpers.ServerError(e, nil)
29 }
30
31 bs := s.getBlockstore(urepo.Repo.Did)
32
33 cs, err := car.NewCarReader(bytes.NewReader(b))
34 if err != nil {
35 logger.Error("could not read car in import request", "error", err)
36 return helpers.ServerError(e, nil)
37 }
38
39 orderedBlocks := []blocks.Block{}
40 currBlock, err := cs.Next()
41 if err != nil {
42 logger.Error("could not get first block from car", "error", err)
43 return helpers.ServerError(e, nil)
44 }
45 currBlockCt := 1
46
47 for currBlock != nil {
48 logger.Info("someone is importing their repo", "block", currBlockCt)
49 orderedBlocks = append(orderedBlocks, currBlock)
50 next, _ := cs.Next()
51 currBlock = next
52 currBlockCt++
53 }
54
55 slices.Reverse(orderedBlocks)
56
57 if err := bs.PutMany(context.TODO(), orderedBlocks); err != nil {
58 logger.Error("could not insert blocks", "error", err)
59 return helpers.ServerError(e, nil)
60 }
61
62 r, err := openRepo(context.TODO(), bs, cs.Header.Roots[0], urepo.Repo.Did)
63 if err != nil {
64 logger.Error("could not open repo", "error", err)
65 return helpers.ServerError(e, nil)
66 }
67
68 tx := s.db.Begin(ctx)
69
70 clock := syntax.NewTIDClock(0)
71
72 if err := r.MST.Walk(func(key []byte, cid cid.Cid) error {
73 pts := strings.Split(string(key), "/")
74 nsid := pts[0]
75 rkey := pts[1]
76 cidStr := cid.String()
77 b, err := bs.Get(context.TODO(), cid)
78 if err != nil {
79 logger.Error("record bytes don't exist in blockstore", "error", err)
80 return helpers.ServerError(e, nil)
81 }
82
83 rec := models.Record{
84 Did: urepo.Repo.Did,
85 CreatedAt: clock.Next().String(),
86 Nsid: nsid,
87 Rkey: rkey,
88 Cid: cidStr,
89 Value: b.RawData(),
90 }
91
92 if err := tx.Save(rec).Error; err != nil {
93 return err
94 }
95
96 return nil
97 }); err != nil {
98 tx.Rollback()
99 logger.Error("record bytes don't exist in blockstore", "error", err)
100 return helpers.ServerError(e, nil)
101 }
102
103 tx.Commit()
104
105 root, rev, err := commitRepo(context.TODO(), bs, r, urepo.Repo.SigningKey)
106 if err != nil {
107 logger.Error("error committing", "error", err)
108 return helpers.ServerError(e, nil)
109 }
110
111 if err := s.UpdateRepo(context.TODO(), urepo.Repo.Did, root, rev); err != nil {
112 logger.Error("error updating repo after commit", "error", err)
113 return helpers.ServerError(e, nil)
114 }
115
116 return nil
117}