1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package gitcli
4
5import (
6 "bufio"
7 "bytes"
8 "context"
9 "errors"
10 "fmt"
11 "io"
12 "net/http"
13 "os"
14 "os/exec"
15 "path"
16 "strings"
17 "sync"
18
19 "bnema.dev/ermite/internal/domain"
20)
21
22// smallBlob is the largest blob read whole, through the long-lived cat-file
23// process. A larger one streams from a process of its own, so that a slow
24// reader never holds the shared process.
25const smallBlob = 256 << 10
26
27// OpenBlob describes the file at path and returns its content as a stream.
28//
29// A blob up to 256 KiB is read through the repository's long-lived cat-file
30// process. A larger one gets a `git cat-file blob` process that lives until
31// the reader is closed, or ctx ends, or the adapter is closed; the memory
32// used is a few buffers, whatever the size. The reader fails with
33// io.ErrUnexpectedEOF if git stops before the announced size.
34//
35//hexcheck:ignore suspicious-business-logic-in-adapter builds domain values from git output; size and binary rules are the port contract
36func (g *Git) OpenBlob(ctx context.Context, repo domain.RepoName, rev domain.Revision, p domain.TreePath) (domain.BlobInfo, io.ReadCloser, error) {
37 dir, err := g.repoDir(repo)
38 if err != nil {
39 return domain.BlobInfo{}, nil, err
40 }
41 if p.IsRoot() {
42 return domain.BlobInfo{}, nil, fmt.Errorf("blob of %q: %w", repo, domain.ErrNotFile)
43 }
44 var (
45 h catInfo
46 data []byte
47 )
48 err = g.cat.do(ctx, dir, func(c *catConn) error {
49 h, data = catInfo{}, nil
50 res, err := resolve(c, rev)
51 if err != nil {
52 return err
53 }
54 h, err = c.info(res.CommitID.String() + ":" + p.String())
55 if notFound(err) {
56 return fmt.Errorf("path %q: %w", p, domain.ErrNotFound)
57 }
58 if err != nil {
59 return err
60 }
61 if h.Type != "blob" || h.Size > smallBlob {
62 return nil
63 }
64 _, data, err = c.contents(h.ID, smallBlob)
65 return err
66 })
67 if err != nil {
68 return domain.BlobInfo{}, nil, fmt.Errorf("blob %q of %q: %w", p, repo, err)
69 }
70 if h.Type != "blob" {
71 return domain.BlobInfo{}, nil, fmt.Errorf("blob %q of %q: %w", p, repo, domain.ErrNotFile)
72 }
73 id, err := domain.ParseObjectID(h.ID)
74 if err != nil {
75 return domain.BlobInfo{}, nil, err
76 }
77 info := domain.BlobInfo{ID: id, Path: p.String(), Size: h.Size, Mode: h.Mode}
78
79 if data != nil || h.Size == 0 {
80 fillMedia(&info, data)
81 return info, io.NopCloser(bytes.NewReader(data)), nil
82 }
83
84 s, err := g.startBlobStream(ctx, dir, id, h.Size)
85 if err != nil {
86 return domain.BlobInfo{}, nil, fmt.Errorf("blob %q of %q: %w", p, repo, err)
87 }
88 head, err := s.r.Peek(min(int(h.Size), domain.BinarySniffLen))
89 if err != nil && !errors.Is(err, bufio.ErrBufferFull) {
90 _ = s.Close()
91 return domain.BlobInfo{}, nil, fmt.Errorf("blob %q of %q: %w", p, repo, ctxOr(ctx, err))
92 }
93 fillMedia(&info, head)
94 return info, s, nil
95}
96
97func ctxOr(ctx context.Context, err error) error {
98 if cerr := ctx.Err(); cerr != nil {
99 return cerr
100 }
101 return err
102}
103
104// fillMedia sets Binary and MediaType from the first bytes of the content.
105func fillMedia(info *domain.BlobInfo, content []byte) {
106 head := content[:min(len(content), domain.BinarySniffLen)]
107 info.Binary = bytes.IndexByte(head, 0) >= 0
108 info.MediaType = mediaType(info.Path, head, info.Binary)
109}
110
111// textTypes maps the extensions whose media type is more precise than what
112// the content shows. It is fixed here so that the answer does not depend on the
113// MIME tables of the host.
114var textTypes = map[string]string{
115 ".html": "text/html", ".htm": "text/html", ".css": "text/css",
116 ".js": "text/javascript", ".mjs": "text/javascript",
117 ".json": "application/json", ".xml": "application/xml",
118 ".svg": "image/svg+xml", ".md": "text/markdown", ".csv": "text/csv",
119}
120
121// mediaType guesses a media type without parameters. What the content shows
122// (images, archives, PDF…) wins; for plain text the file name may refine it
123// from textTypes, but a name never makes binary content text.
124func mediaType(name string, head []byte, binary bool) string {
125 sniffed, _, _ := strings.Cut(http.DetectContentType(head), ";")
126 switch {
127 case binary && strings.HasPrefix(sniffed, "text/"):
128 return "application/octet-stream"
129 case sniffed != "text/plain" && sniffed != "text/xml" && sniffed != "application/octet-stream":
130 return sniffed
131 case binary:
132 return "application/octet-stream"
133 }
134 if t, ok := textTypes[strings.ToLower(path.Ext(name))]; ok {
135 return t
136 }
137 return "text/plain"
138}
139
140// blobStream is the content of a blob read from its own git process.
141type blobStream struct {
142 pool *catPool
143 cmd *exec.Cmd
144 pipe *os.File
145 r *bufio.Reader
146 size int64
147 read int64
148 stop func() bool // detaches the ctx watcher
149
150 done chan struct{} // closed when the process has been collected
151 once sync.Once
152 release func() // gives the stream slot back
153}
154
155// startBlobStream starts the process of a blob stream. The process is counted
156// against Config.MaxBlobStreams from its start until the stream is closed (or
157// the process is collected); there is no queue: a full table is refused with
158// domain.ErrResourceExhausted.
159func (g *Git) startBlobStream(ctx context.Context, dir string, id domain.ObjectID, size int64) (*blobStream, error) {
160 select {
161 case g.blobs <- struct{}{}:
162 default:
163 return nil, fmt.Errorf("too many blob streams open: %w", domain.ErrResourceExhausted)
164 }
165 release := func() { <-g.blobs }
166 pr, pw, err := os.Pipe()
167 if err != nil {
168 release()
169 return nil, fmt.Errorf("start git cat-file: %w", err)
170 }
171 cmd := exec.Command(g.bin, readArgs(dir, "cat-file", "blob", id.String())...)
172 cmd.Env = g.env
173 cmd.Dir = g.dataDir
174 cmd.Stdout = pw
175 if err := cmd.Start(); err != nil {
176 _ = pr.Close()
177 _ = pw.Close()
178 release()
179 return nil, fmt.Errorf("start git cat-file: %w", err)
180 }
181 _ = pw.Close()
182 s := &blobStream{
183 pool: g.cat, cmd: cmd, pipe: pr, r: bufio.NewReaderSize(pr, 64*1024),
184 size: size, done: make(chan struct{}), release: release,
185 }
186 go s.reap()
187 if err := g.cat.addStream(s); err != nil {
188 _ = s.Close()
189 return nil, err
190 }
191 s.stop = context.AfterFunc(ctx, s.kill)
192 return s, nil
193}
194
195// reap collects the process and, with it, the stream slot: a process that is
196// gone no longer counts, whether or not the reader was closed.
197func (s *blobStream) reap() {
198 _ = s.cmd.Wait()
199 s.release()
200 close(s.done)
201}
202
203func (s *blobStream) kill() { _ = s.cmd.Process.Kill() }
204
205// Read implements io.Reader.
206func (s *blobStream) Read(p []byte) (int, error) {
207 n, err := s.r.Read(p)
208 s.read += int64(n)
209 if errors.Is(err, io.EOF) && s.read < s.size {
210 return n, io.ErrUnexpectedEOF
211 }
212 return n, err
213}
214
215// Close stops the process, waits for it and releases the pipe. It is safe to
216// call more than once.
217func (s *blobStream) Close() error {
218 s.once.Do(func() {
219 if s.stop != nil {
220 s.stop()
221 }
222 s.kill()
223 <-s.done
224 _ = s.pipe.Close()
225 s.pool.removeStream(s)
226 })
227 return nil
228}