@@ -11,6 +11,8 @@ import (
1111 "net/http"
1212 "net/url"
1313 "os"
14+ "strconv"
15+ "syscall"
1416 "time"
1517)
1618
@@ -55,10 +57,10 @@ func NewTursoServerClient(baseURL *url.URL, token string, cliVersion string, org
5557 }, nil
5658}
5759
58- // UploadFile uploads a database file to the Turso server.
60+ // UploadFileSinglePart uploads a database file to the Turso server using a single request .
5961// it assumes a SQLite file exists at 'filepath'.
6062// it streams the file to the server, and calls the onProgress callback with the progress of the upload.
61- func (i * TursoServerClient ) UploadFile (filepath , remoteEncryptionCipher , remoteEncryptionKey string , onUploadProgress func (progressPct int , uploadedBytes int64 , totalBytes int64 , elapsedTime time.Duration , done bool )) error {
63+ func (i * TursoServerClient ) UploadFileSinglePart (filepath , remoteEncryptionCipher , remoteEncryptionKey string , onUploadProgress func (progressPct int , uploadedBytes int64 , totalBytes int64 , elapsedTime time.Duration , done bool )) error {
6264 file , err := os .Open (filepath )
6365 if err != nil {
6466 return fmt .Errorf ("failed to open file %s: %w" , filepath , err )
@@ -105,6 +107,147 @@ func (i *TursoServerClient) UploadFile(filepath, remoteEncryptionCipher, remoteE
105107 return nil
106108}
107109
110+ // UploadFileMultipart uploads a database file using the multipart upload flow.
111+ func (i * TursoServerClient ) UploadFileMultipart (filepath string , remoteEncryptionCipher , remoteEncryptionKey string , onUploadProgress func (progressPct int , uploadedBytes int64 , totalBytes int64 , elapsedTime time.Duration , done bool )) error {
112+ file , err := os .Open (filepath )
113+ if err != nil {
114+ return fmt .Errorf ("failed to open file %s: %w" , filepath , err )
115+ }
116+ defer file .Close ()
117+
118+ if err := syscall .Flock (int (file .Fd ()), syscall .LOCK_EX ); err == nil {
119+ // locking is on a best effort basis
120+ defer syscall .Flock (int (file .Fd ()), syscall .LOCK_UN )
121+ }
122+
123+ stat , err := file .Stat ()
124+ if err != nil {
125+ return fmt .Errorf ("failed to get file stats for %s: %w" , filepath , err )
126+ }
127+
128+ totalSize := stat .Size ()
129+ startTime := time .Now ()
130+
131+ chunkSize , err := i .startMultipartUpload (totalSize )
132+ if err != nil {
133+ return err
134+ }
135+
136+ uploadedBytes , err := i .uploadChunks (chunkSize , file , totalSize , startTime , remoteEncryptionCipher , remoteEncryptionKey , onUploadProgress )
137+ if err != nil {
138+ return err
139+ }
140+
141+ if err = i .finalizeUpload (); err != nil {
142+ return err
143+ }
144+
145+ elapsedTime := time .Since (startTime )
146+ onUploadProgress (100 , uploadedBytes , totalSize , elapsedTime , true )
147+
148+ return nil
149+ }
150+
151+ func (i * TursoServerClient ) startMultipartUpload (dbSize int64 ) (int64 , error ) {
152+ requestBody := map [string ]int64 {
153+ "db_size_bytes" : dbSize ,
154+ }
155+
156+ body , err := marshal (requestBody )
157+ if err != nil {
158+ return 0 , fmt .Errorf ("failed to marshal multipart upload request: %w" , err )
159+ }
160+
161+ r , err := i .client .Put ("/v2/upload/start" , body )
162+ if err != nil {
163+ return 0 , fmt .Errorf ("failed to initiate multipart upload: %w" , err )
164+ }
165+ defer r .Body .Close ()
166+
167+ if r .StatusCode != http .StatusOK {
168+ body , err := io .ReadAll (r .Body )
169+ if err != nil {
170+ return 0 , fmt .Errorf ("initiate multipart upload failed with status code %d and error reading response: %v" , r .StatusCode , err )
171+ }
172+ return 0 , fmt .Errorf ("initiate multipart upload failed with status code %d: %s" , r .StatusCode , string (body ))
173+ }
174+
175+ type multipartUploadResponse struct {
176+ ChunkSize int64 `json:"chunk_size"`
177+ }
178+ var uploadResp multipartUploadResponse
179+ if err := json .NewDecoder (r .Body ).Decode (& uploadResp ); err != nil {
180+ return 0 , fmt .Errorf ("failed to decode multipart upload response: %w" , err )
181+ }
182+
183+ return uploadResp .ChunkSize , nil
184+ }
185+
186+ func (i * TursoServerClient ) uploadChunks (chunkSize int64 , file io.Reader , totalSize int64 , startTime time.Time , remoteEncryptionCipher , remoteEncryptionKey string , onUploadProgress func (progressPct int , uploadedBytes int64 , totalBytes int64 , elapsedTime time.Duration , done bool )) (int64 , error ) {
187+ var uploadedBytes int64 = 0
188+ chunkID := 0
189+
190+ for uploadedBytes < totalSize {
191+ remaining := totalSize - uploadedBytes
192+ currentChunkSize := chunkSize
193+ if remaining < chunkSize {
194+ currentChunkSize = remaining
195+ }
196+
197+ chunkReader := io .LimitReader (file , currentChunkSize )
198+ chunkPath := fmt .Sprintf ("/v2/upload/chunk/%d" , chunkID )
199+
200+ var headers = map [string ]string {}
201+ if remoteEncryptionCipher != "" && remoteEncryptionKey != "" {
202+ headers [EncryptionCipherHeader ] = remoteEncryptionCipher
203+ headers [EncryptionKeyHeader ] = remoteEncryptionKey
204+ }
205+ headers ["Content-Length" ] = strconv .FormatInt (currentChunkSize , 10 )
206+
207+ r , err := i .client .PutBinary (chunkPath , chunkReader , headers )
208+ if err != nil {
209+ return 0 , fmt .Errorf ("failed to upload chunk %d: %w" , chunkID , err )
210+ }
211+
212+ if r .StatusCode != http .StatusOK && r .StatusCode != http .StatusCreated {
213+ if body , err := io .ReadAll (r .Body ); err != nil {
214+ _ = r .Body .Close ()
215+ return 0 , fmt .Errorf ("upload chunk %d failed with status code %d and error reading response: %v" , chunkID , r .StatusCode , err )
216+ } else {
217+ _ = r .Body .Close ()
218+ return 0 , fmt .Errorf ("upload chunk %d failed with status code %d: %s" , chunkID , r .StatusCode , string (body ))
219+ }
220+ } else {
221+ _ = r .Body .Close ()
222+ }
223+
224+ uploadedBytes += currentChunkSize
225+ progressPct := int (float64 (uploadedBytes ) / float64 (totalSize ) * 100 )
226+ elapsedTime := time .Since (startTime )
227+ onUploadProgress (progressPct , uploadedBytes , totalSize , elapsedTime , false )
228+
229+ chunkID ++
230+ }
231+ return uploadedBytes , nil
232+ }
233+
234+ func (i * TursoServerClient ) finalizeUpload () error {
235+ r , err := i .client .Put ("/v2/upload/finalize" , nil )
236+ if err != nil {
237+ return fmt .Errorf ("failed to finalize multipart upload: %w" , err )
238+ }
239+ defer r .Body .Close ()
240+
241+ if r .StatusCode != http .StatusOK {
242+ body , err := io .ReadAll (r .Body )
243+ if err != nil {
244+ return fmt .Errorf ("finalize multipart upload failed with status code %d and error reading response: %v" , r .StatusCode , err )
245+ }
246+ return fmt .Errorf ("finalize multipart upload failed with status code %d: %s" , r .StatusCode , string (body ))
247+ }
248+ return nil
249+ }
250+
108251type ExportInfo struct {
109252 CurrentGeneration int `json:"current_generation"`
110253}
0 commit comments