diff --git a/backend/_example/memory_store/accessor/image.go b/backend/_example/memory_store/accessor/image.go
index 6f4b385c..e667ec9b 100644
--- a/backend/_example/memory_store/accessor/image.go
+++ b/backend/_example/memory_store/accessor/image.go
@@ -13,6 +13,8 @@ import (
log "github.com/go-pkgz/lgr"
"github.com/pkg/errors"
+
+ "github.com/umputun/remark/backend/app/store/image"
)
// MemImage implements image.Store with memory backend
@@ -95,3 +97,17 @@ func (m *MemImage) Cleanup(_ context.Context, ttl time.Duration) error {
m.Unlock()
return nil
}
+
+// Info returns meta information about storage
+func (m *MemImage) Info() (image.StoreInfo, error) {
+ var ts time.Time
+ m.RLock()
+ for _, t := range m.insertTime {
+ if ts.IsZero() || t.Before(ts) {
+ ts = t
+ }
+ }
+ m.RUnlock()
+
+ return image.StoreInfo{FirstStagingImageTS: ts}, nil
+}
diff --git a/backend/_example/memory_store/accessor/image_test.go b/backend/_example/memory_store/accessor/image_test.go
index bf496d0b..92679c63 100644
--- a/backend/_example/memory_store/accessor/image_test.go
+++ b/backend/_example/memory_store/accessor/image_test.go
@@ -80,3 +80,23 @@ func TestMemImage_Cleanup(t *testing.T) {
err := svc.Cleanup(context.TODO(), time.Minute)
assert.NoError(t, err)
}
+
+func TestMemImage_Info(t *testing.T) {
+ svc := NewMemImageStore()
+ gopher, err := ioutil.ReadAll(gopherPNG())
+ assert.NoError(t, err)
+
+ // get info on empty storage, should be zero
+ info, err := svc.Info()
+ assert.NoError(t, err)
+ assert.True(t, info.FirstStagingImageTS.IsZero())
+
+ // save image
+ err = svc.Save("test_img", gopher)
+ assert.NoError(t, err)
+
+ // get info after saving, should be non-zero
+ info, err = svc.Info()
+ assert.NoError(t, err)
+ assert.False(t, info.FirstStagingImageTS.IsZero())
+}
diff --git a/backend/_example/memory_store/server/image.go b/backend/_example/memory_store/server/image.go
index 9b550206..ab90bd9d 100644
--- a/backend/_example/memory_store/server/image.go
+++ b/backend/_example/memory_store/server/image.go
@@ -54,3 +54,8 @@ func (s *RPC) imgCleanupHndl(id uint64, params json.RawMessage) (rr jrpc.Respons
err := s.img.Cleanup(context.TODO(), ttl)
return jrpc.EncodeResponse(id, nil, err)
}
+
+func (s *RPC) imgInfoHndl(id uint64, _ json.RawMessage) (rr jrpc.Response) {
+ info, err := s.img.Info()
+ return jrpc.EncodeResponse(id, info, err)
+}
diff --git a/backend/_example/memory_store/server/image_test.go b/backend/_example/memory_store/server/image_test.go
index d264f3ec..0a5ea027 100644
--- a/backend/_example/memory_store/server/image_test.go
+++ b/backend/_example/memory_store/server/image_test.go
@@ -124,3 +124,25 @@ func TestRPC_imgCleanupHndl(t *testing.T) {
_, err = ri.Load(id)
assert.EqualError(t, err, "image test_img not found")
}
+
+func TestRPC_imgInfoHndl(t *testing.T) {
+ port, teardown := prepTestStore(t)
+ defer teardown()
+ api := fmt.Sprintf("http://localhost:%d/test", port)
+
+ ri := image.RPC{Client: jrpc.Client{API: api, Client: http.Client{Timeout: 1 * time.Second}}}
+
+ // get info on empty storage, should be zero
+ info, err := ri.Info()
+ assert.NoError(t, err)
+ assert.True(t, info.FirstStagingImageTS.IsZero())
+
+ // save
+ err = ri.Save("test_img", gopherPNGBytes())
+ assert.NoError(t, err)
+
+ // get info after saving, should be non-zero
+ info, err = ri.Info()
+ assert.NoError(t, err)
+ assert.False(t, info.FirstStagingImageTS.IsZero())
+}
diff --git a/backend/_example/memory_store/server/rpc.go b/backend/_example/memory_store/server/rpc.go
index 8dbe6568..07609e7d 100644
--- a/backend/_example/memory_store/server/rpc.go
+++ b/backend/_example/memory_store/server/rpc.go
@@ -61,5 +61,6 @@ func (s *RPC) addHandlers() {
"load": s.imgLoadHndl,
"commit": s.imgCommitHndl,
"cleanup": s.imgCleanupHndl,
+ "info": s.imgInfoHndl,
})
}
diff --git a/backend/app/cmd/server.go b/backend/app/cmd/server.go
index 54d0fb7c..220982ef 100644
--- a/backend/app/cmd/server.go
+++ b/backend/app/cmd/server.go
@@ -338,7 +338,7 @@ func (s *ServerCommand) newServerApp() (*serverApp, error) {
if err != nil {
return nil, errors.Wrap(err, "failed to make pictures store")
}
- log.Printf("[DEBUG] image service for url=%s, ttl=%v", imageService.ImageAPI, imageService.TTL)
+ log.Printf("[DEBUG] image service for url=%s, EditDuration=%v", imageService.ImageAPI, imageService.EditDuration)
dataService := &service.DataStore{
Engine: storeEngine,
@@ -487,6 +487,11 @@ func (a *serverApp) run(ctx context.Context) error {
go a.devAuth.Run(context.Background()) // dev oauth2 server on :8084
}
+ // staging images resubmit after restart of the app
+ if e := a.dataService.ResubmitStagingImages(a.Sites); e != nil {
+ log.Printf("[WARN] failed to resubmit comments with staging images, %s", e)
+ }
+
go a.imageService.Cleanup(ctx) // pictures cleanup for staging images
a.restSrv.Run(a.Port)
@@ -580,11 +585,11 @@ func (s *ServerCommand) makeAvatarStore() (avatar.Store, error) {
func (s *ServerCommand) makePicturesStore() (*image.Service, error) {
imageServiceParams := image.ServiceParams{
- ImageAPI: s.RemarkURL + "/api/v1/picture/",
- TTL: 5 * s.EditDuration, // add extra time to image TTL for staging
- MaxSize: s.Image.MaxSize,
- MaxHeight: s.Image.ResizeHeight,
- MaxWidth: s.Image.ResizeWidth,
+ ImageAPI: s.RemarkURL + "/api/v1/picture/",
+ EditDuration: s.EditDuration,
+ MaxSize: s.Image.MaxSize,
+ MaxHeight: s.Image.ResizeHeight,
+ MaxWidth: s.Image.ResizeWidth,
}
switch s.Image.Type {
case "bolt":
diff --git a/backend/app/rest/api/rest_private_test.go b/backend/app/rest/api/rest_private_test.go
index 39d45189..729c6d05 100644
--- a/backend/app/rest/api/rest_private_test.go
+++ b/backend/app/rest/api/rest_private_test.go
@@ -908,8 +908,8 @@ func TestRest_CreateWithPictures(t *testing.T) {
Staging: "/tmp/remark42/images.staging",
Location: "/tmp/remark42/images",
}, image.ServiceParams{
- TTL: 100 * time.Millisecond,
- MaxSize: 2000,
+ EditDuration: 100 * time.Millisecond,
+ MaxSize: 2000,
})
svc.privRest.imageService = imageService
diff --git a/backend/app/rest/api/rest_test.go b/backend/app/rest/api/rest_test.go
index cf1f2c1d..2b4b5366 100644
--- a/backend/app/rest/api/rest_test.go
+++ b/backend/app/rest/api/rest_test.go
@@ -376,8 +376,8 @@ func startupT(t *testing.T) (ts *httptest.Server, srv *Rest, teardown func()) {
Partitions: 100,
Staging: tmp + "/pics-remark42/staging",
}, image.ServiceParams{
- TTL: 100 * time.Millisecond,
- MaxSize: 10000,
+ EditDuration: 100 * time.Millisecond,
+ MaxSize: 10000,
}),
ImageProxy: &proxy.Image{},
ReadOnlyAge: 10,
diff --git a/backend/app/store/image/bolt_store.go b/backend/app/store/image/bolt_store.go
index 5ed8d68c..ff1ac801 100644
--- a/backend/app/store/image/bolt_store.go
+++ b/backend/app/store/image/bolt_store.go
@@ -140,3 +140,26 @@ func (b *Bolt) Cleanup(_ context.Context, ttl time.Duration) error {
})
return err
}
+
+// Info returns meta information about storage
+func (b *Bolt) Info() (StoreInfo, error) {
+ var ts time.Time
+ err := b.db.View(func(tx *bolt.Tx) error {
+ c := tx.Bucket([]byte(insertTimeBktName)).Cursor()
+
+ for id, tsData := c.First(); id != nil; id, tsData = c.Next() {
+ var createdRaw int64
+ err := binary.Read(bytes.NewReader(tsData), binary.LittleEndian, &createdRaw)
+ if err != nil {
+ return errors.Wrapf(err, "failed to deserialize timestamp for %s", id)
+ }
+
+ created := time.Unix(0, createdRaw)
+ if ts.IsZero() || created.Before(ts) {
+ ts = created
+ }
+ }
+ return nil
+ })
+ return StoreInfo{FirstStagingImageTS: ts}, errors.Wrapf(err, "problem retrieving first timestamp from staging images")
+}
diff --git a/backend/app/store/image/bolt_store_test.go b/backend/app/store/image/bolt_store_test.go
index 2dce4cc5..14c7dc88 100644
--- a/backend/app/store/image/bolt_store_test.go
+++ b/backend/app/store/image/bolt_store_test.go
@@ -105,6 +105,25 @@ func TestBoltStore_Cleanup(t *testing.T) {
assert.NoError(t, err)
}
+func TestBolt_Info(t *testing.T) {
+ svc, teardown := prepareBoltImageStorageTest(t)
+ defer teardown()
+
+ // get info on empty storage, should be zero
+ info, err := svc.Info()
+ assert.NoError(t, err)
+ assert.True(t, info.FirstStagingImageTS.IsZero())
+
+ // save image
+ err = svc.Save("test_img", gopherPNGBytes())
+ assert.NoError(t, err)
+
+ // get info after saving, should be non-zero
+ info, err = svc.Info()
+ assert.NoError(t, err)
+ assert.False(t, info.FirstStagingImageTS.IsZero())
+}
+
func assertBoltImgNil(t *testing.T, db *bolt.DB, bucket string, id string) {
checkBoltImgData(t, db, bucket, id, func(data []byte) error {
assert.Nil(t, data, id)
diff --git a/backend/app/store/image/fs_store.go b/backend/app/store/image/fs_store.go
index 420dde59..5306dc35 100644
--- a/backend/app/store/image/fs_store.go
+++ b/backend/app/store/image/fs_store.go
@@ -115,6 +115,33 @@ func (f *FileSystem) Cleanup(_ context.Context, ttl time.Duration) error {
return errors.Wrap(err, "failed to cleanup images")
}
+// Info returns meta information about storage
+func (f *FileSystem) Info() (StoreInfo, error) {
+ if _, err := os.Stat(f.Staging); os.IsNotExist(err) {
+ return StoreInfo{}, nil
+ }
+
+ var ts time.Time
+ err := filepath.Walk(f.Staging, func(fpath string, info os.FileInfo, err error) error {
+ if err != nil {
+ return err
+ }
+ if info.IsDir() {
+ return nil
+ }
+
+ created := info.ModTime()
+ if ts.IsZero() || created.Before(ts) {
+ ts = created
+ }
+ return nil
+ })
+ if err != nil {
+ return StoreInfo{}, errors.Wrapf(err, "problem retrieving first timestamp from staging images on fs")
+ }
+ return StoreInfo{FirstStagingImageTS: ts}, nil
+}
+
// location gets full path for id by adding partition to the final path in order to keep files in different subdirectories
// and avoid too many files in a single place.
// the end result is a full path like this - /tmp/images/user1/92/xxx-yyy.png.
diff --git a/backend/app/store/image/fs_store_test.go b/backend/app/store/image/fs_store_test.go
index 17668cce..9ac57a44 100644
--- a/backend/app/store/image/fs_store_test.go
+++ b/backend/app/store/image/fs_store_test.go
@@ -107,7 +107,6 @@ func TestFsStore_LoadAfterSave(t *testing.T) {
id := "test_img"
err := svc.Save(id, gopherPNGBytes())
assert.NoError(t, err)
- t.Log(id)
data, err := svc.Load(id)
assert.NoError(t, err)
@@ -125,7 +124,6 @@ func TestFsStore_LoadAfterCommit(t *testing.T) {
id := "test_img"
err := svc.Save(id, gopherPNGBytes())
assert.NoError(t, err)
- t.Log(id)
err = svc.Commit(id)
require.NoError(t, err)
@@ -233,6 +231,25 @@ func TestFsStore_Cleanup(t *testing.T) {
assert.Error(t, err, "no file on staging anymore")
}
+func TestFsStore_Info(t *testing.T) {
+ svc, teardown := prepareImageTest(t)
+ defer teardown()
+
+ // get ts on empty storage, should be zero
+ ts, err := svc.Info()
+ assert.NoError(t, err)
+ assert.True(t, ts.FirstStagingImageTS.IsZero())
+
+ // save image
+ err = svc.Save("test_img", gopherPNGBytes())
+ assert.NoError(t, err)
+
+ // get ts after saving, should be non-zero
+ ts, err = svc.Info()
+ assert.NoError(t, err)
+ assert.False(t, ts.FirstStagingImageTS.IsZero())
+}
+
func prepareImageTest(t *testing.T) (svc *FileSystem, teardown func()) {
loc, err := ioutil.TempDir("", "test_image_r42")
require.NoError(t, err, "failed to make temp dir")
diff --git a/backend/app/store/image/image.go b/backend/app/store/image/image.go
index e121aee2..2f5cc7b9 100644
--- a/backend/app/store/image/image.go
+++ b/backend/app/store/image/image.go
@@ -31,7 +31,7 @@ import (
// Service wraps Store with common functions needed for any store implementation
// It also provides async Submit with func param retrieving all submitting ids.
-// Submitted ids committed (i.e. moved from staging to final) on TTL expiration.
+// Submitted ids committed (i.e. moved from staging to final) on commitTTL expiration.
type Service struct {
ServiceParams
@@ -45,11 +45,22 @@ type Service struct {
// ServiceParams contains externally adjustable parameters of Service
type ServiceParams struct {
- TTL time.Duration // for how long file allowed on staging
- ImageAPI string // image api matching path
- MaxSize int
- MaxHeight int
- MaxWidth int
+ EditDuration time.Duration // edit period for comments
+ ImageAPI string // image api matching path
+ MaxSize int
+ MaxHeight int
+ MaxWidth int
+
+ // duration of time after which images are checked and committed if still
+ // present in the submitted comment after it's EditDuration is expired
+ commitTTL time.Duration
+ // duration of time after which images are deleted from staging
+ cleanupTTL time.Duration
+}
+
+// StoreInfo contains image store meta information
+type StoreInfo struct {
+ FirstStagingImageTS time.Time
}
// To regenerate mock run from this directory:
@@ -60,6 +71,7 @@ type ServiceParams struct {
// Two-stage commit scheme is used for not storing images which are uploaded but later never used in the comments,
// e.g. when somebody uploaded a picture but did not sent the comment.
type Store interface {
+ Info() (StoreInfo, error) // get meta information about storage
Save(id string, img []byte) error // store image with passed id to staging
Load(id string) ([]byte, error) // load image by ID
@@ -76,6 +88,10 @@ type submitReq struct {
// NewService returns new Service instance
func NewService(s Store, p ServiceParams) *Service {
+ p.commitTTL = p.EditDuration * 15 / 10 // Commit call on every 1.5 * EditDuration
+ p.cleanupTTL = p.EditDuration * 25 / 10 // Cleanup call on every 2.5 * EditDuration
+ // In case Cleanup and Submit start at the same time (case of stale staging images check
+ // on the program start) these TTL values guarantee that Commit will happen before Cleanup.
return &Service{ServiceParams: p, store: s}
}
@@ -92,8 +108,8 @@ func (s *Service) Submit(idsFn func() []string) {
go func() {
defer s.wg.Done()
for req := range s.submitCh {
- // wait for TTL expiration with emergency pass on term
- for atomic.LoadInt32(&s.term) == 0 && time.Since(req.TS) <= s.TTL/2 { // commit on a half of TTL
+ // wait for commitTTL expiration with emergency pass on term
+ for atomic.LoadInt32(&s.term) == 0 && time.Since(req.TS) <= s.commitTTL {
time.Sleep(time.Millisecond * 10) // small sleep to relive busy wait but keep reactive for term (close)
}
for _, id := range req.idsFn() {
@@ -133,23 +149,28 @@ func (s *Service) ExtractPictures(commentHTML string) (ids []string, err error)
return ids, nil
}
-// Cleanup runs periodic cleanup with TTL. Blocking loop, should be called inside of goroutine by consumer
+// Cleanup runs periodic cleanup with cleanupTTL. Blocking loop, should be called inside of goroutine by consumer
func (s *Service) Cleanup(ctx context.Context) {
- log.Printf("[INFO] start pictures cleanup, staging ttl=%v", s.TTL)
+ log.Printf("[INFO] start pictures cleanup, staging ttl=%v", s.cleanupTTL)
for {
select {
case <-ctx.Done():
log.Printf("[INFO] cleanup terminated, %v", ctx.Err())
return
- case <-time.After(s.TTL / 2): // cleanup call on every 1/2 TTL
- if err := s.store.Cleanup(ctx, s.TTL); err != nil {
+ case <-time.After(s.cleanupTTL):
+ if err := s.store.Cleanup(ctx, s.cleanupTTL); err != nil {
log.Printf("[WARN] failed to cleanup, %v", err)
}
}
}
}
+// Info returns meta information about storage
+func (s *Service) Info() (StoreInfo, error) {
+ return s.store.Info()
+}
+
// Close flushes all in-progress submits and enforces waiting commits
func (s *Service) Close(ctx context.Context) {
log.Printf("[INFO] close image service ")
diff --git a/backend/app/store/image/image_mock.go b/backend/app/store/image/image_mock.go
index 25e6775d..f9c74ee4 100644
--- a/backend/app/store/image/image_mock.go
+++ b/backend/app/store/image/image_mock.go
@@ -39,6 +39,27 @@ func (_m *MockStore) Commit(id string) error {
return r0
}
+// Info provides a mock function with given fields:
+func (_m *MockStore) Info() (StoreInfo, error) {
+ ret := _m.Called()
+
+ var r0 StoreInfo
+ if rf, ok := ret.Get(0).(func() StoreInfo); ok {
+ r0 = rf()
+ } else {
+ r0 = ret.Get(0).(StoreInfo)
+ }
+
+ var r1 error
+ if rf, ok := ret.Get(1).(func() error); ok {
+ r1 = rf()
+ } else {
+ r1 = ret.Error(1)
+ }
+
+ return r0, r1
+}
+
// Load provides a mock function with given fields: id
func (_m *MockStore) Load(id string) ([]byte, error) {
ret := _m.Called(id)
diff --git a/backend/app/store/image/image_test.go b/backend/app/store/image/image_test.go
index 228ac45e..cd77995f 100644
--- a/backend/app/store/image/image_test.go
+++ b/backend/app/store/image/image_test.go
@@ -96,7 +96,7 @@ func TestService_Cleanup(t *testing.T) {
store := MockStore{}
store.On("Cleanup", mock.Anything, mock.Anything).Times(10).Return(nil)
- svc := Service{store: &store, ServiceParams: ServiceParams{TTL: 100 * time.Millisecond}}
+ svc := NewService(&store, ServiceParams{EditDuration: 20 * time.Millisecond})
ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*549)
defer cancel()
svc.Cleanup(ctx)
@@ -106,7 +106,7 @@ func TestService_Cleanup(t *testing.T) {
func TestService_Submit(t *testing.T) {
store := MockStore{}
store.On("Commit", mock.Anything, mock.Anything).Times(5).Return(nil)
- svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", TTL: time.Millisecond * 100}}
+ svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", EditDuration: time.Millisecond * 100}}
svc.Submit(func() []string { return []string{"id1", "id2", "id3"} })
svc.Submit(func() []string { return []string{"id4", "id5"} })
svc.Submit(nil)
@@ -119,7 +119,7 @@ func TestService_Submit(t *testing.T) {
func TestService_Close(t *testing.T) {
store := MockStore{}
store.On("Commit", mock.Anything, mock.Anything).Times(5).Return(nil)
- svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", TTL: time.Hour * 24}}
+ svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", EditDuration: time.Hour * 24}}
svc.Submit(func() []string { return []string{"id1", "id2", "id3"} })
svc.Submit(func() []string { return []string{"id4", "id5"} })
svc.Submit(nil)
@@ -130,7 +130,7 @@ func TestService_Close(t *testing.T) {
func TestService_SubmitDelay(t *testing.T) {
store := MockStore{}
store.On("Commit", mock.Anything, mock.Anything).Times(5).Return(nil)
- svc := Service{store: &store, ServiceParams: ServiceParams{ImageAPI: "/blah/", TTL: time.Millisecond * 100}}
+ svc := NewService(&store, ServiceParams{EditDuration: 20 * time.Millisecond})
svc.Submit(func() []string { return []string{"id1", "id2", "id3"} })
time.Sleep(150 * time.Millisecond) // let first batch to pass TTL
svc.Submit(func() []string { return []string{"id4", "id5"} })
@@ -140,6 +140,17 @@ func TestService_SubmitDelay(t *testing.T) {
store.AssertNumberOfCalls(t, "Commit", 5)
}
+func TestService_Info(t *testing.T) {
+ store := MockStore{}
+ store.On("Info", mock.Anything, mock.Anything).Once().Return(StoreInfo{}, nil)
+
+ svc := Service{store: &store, ServiceParams: ServiceParams{}}
+ info, err := svc.Info()
+ assert.NoError(t, err)
+ assert.True(t, info.FirstStagingImageTS.IsZero())
+ store.AssertNumberOfCalls(t, "Info", 1)
+}
+
func TestService_resize(t *testing.T) {
// reader is nil
resized := resize(nil, 100, 100)
diff --git a/backend/app/store/image/remote_store.go b/backend/app/store/image/remote_store.go
index 13a60faa..b0048157 100644
--- a/backend/app/store/image/remote_store.go
+++ b/backend/app/store/image/remote_store.go
@@ -46,3 +46,16 @@ func (r *RPC) Cleanup(_ context.Context, ttl time.Duration) error {
_, err := r.Call("image.cleanup", ttl)
return err
}
+
+// Info returns meta information about storage
+func (r *RPC) Info() (StoreInfo, error) {
+ resp, err := r.Call("image.info")
+ if err != nil {
+ return StoreInfo{}, err
+ }
+ info := StoreInfo{}
+ if err = json.Unmarshal(*resp.Result, &info); err != nil {
+ return StoreInfo{}, err
+ }
+ return info, err
+}
diff --git a/backend/app/store/image/remote_store_test.go b/backend/app/store/image/remote_store_test.go
index e0e214f2..77185b11 100644
--- a/backend/app/store/image/remote_store_test.go
+++ b/backend/app/store/image/remote_store_test.go
@@ -65,6 +65,20 @@ func TestRemote_Cleanup(t *testing.T) {
assert.NoError(t, err)
}
+func TestRemote_Info(t *testing.T) {
+ ts := testServer(t, `{"method":"image.info","id":1}`,
+ `{"result":{"FirstStagingImageTS":"0001-01-01T00:00:01Z"},"id":1}`)
+ defer ts.Close()
+ c := RPC{Client: jrpc.Client{API: ts.URL, Client: http.Client{}}}
+
+ var a Store = &c
+ _ = a
+
+ info, err := c.Info()
+ assert.NoError(t, err)
+ assert.False(t, info.FirstStagingImageTS.IsZero())
+}
+
func testServer(t *testing.T, req, resp string) *httptest.Server {
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
body, err := ioutil.ReadAll(r.Body)
diff --git a/backend/app/store/service/service.go b/backend/app/store/service/service.go
index 4a429d6a..a3d65fd9 100644
--- a/backend/app/store/service/service.go
+++ b/backend/app/store/service/service.go
@@ -200,6 +200,29 @@ func (s *DataStore) DeleteUserDetail(siteID string, userID string, detail engine
})
}
+// ResubmitStagingImages retrieves timestamp of the oldest image in staging and
+// calls s.submitImages on all comments newer than it
+func (s *DataStore) ResubmitStagingImages(sites []string) error {
+ info, err := s.ImageService.Info()
+ if err != nil {
+ return err
+ }
+ ts := info.FirstStagingImageTS
+ if ts.IsZero() {
+ return nil
+ }
+ result := new(multierror.Error)
+ for _, site := range sites {
+ locator := store.Locator{SiteID: site}
+ comments, err := s.FindSince(locator, "time", store.User{}, ts)
+ result = multierror.Append(result, errors.Wrapf(err, "problem finding comments for site %s", site))
+ for _, c := range comments {
+ s.submitImages(c.Locator, c.ID)
+ }
+ }
+ return result.ErrorOrNil()
+}
+
// submitImages initiated delayed commit of all images from the comment uploaded to remark42
func (s *DataStore) submitImages(locator store.Locator, commentID string) {
diff --git a/backend/app/store/service/service_test.go b/backend/app/store/service/service_test.go
index e662b041..1a494bb9 100644
--- a/backend/app/store/service/service_test.go
+++ b/backend/app/store/service/service_test.go
@@ -1,6 +1,7 @@
package service
import (
+ "context"
"fmt"
"io/ioutil"
"math/rand"
@@ -1278,8 +1279,9 @@ func TestService_submitImages(t *testing.T) {
lgr.Setup(lgr.Debug, lgr.CallerFile, lgr.CallerFunc)
mockStore := image.MockStore{}
- mockStore.On("Commit", mock.Anything, mock.Anything).Times(2).Return(nil)
- imgSvc := image.NewService(&mockStore, image.ServiceParams{TTL: 50 * time.Millisecond * 50})
+ mockStore.On("Commit", mock.Anything).Times(2).Return(nil)
+ imgSvc := image.NewService(&mockStore, image.ServiceParams{EditDuration: 50 * time.Millisecond})
+ defer imgSvc.Close(context.TODO())
// two comments for https://radio-t.com
eng, teardown := prepStoreEngine(t)
@@ -1301,6 +1303,105 @@ func TestService_submitImages(t *testing.T) {
time.Sleep(250 * time.Millisecond)
}
+func TestService_ResubmitStagingImages(t *testing.T) {
+ mockStore := image.MockStore{}
+ imgSvc := image.NewService(&mockStore,
+ image.ServiceParams{
+ EditDuration: 10 * time.Millisecond,
+ ImageAPI: "http://127.0.0.1:8080/api/v1/picture/",
+ })
+ defer imgSvc.Close(context.TODO())
+
+ eng, teardown := prepStoreEngine(t)
+ defer teardown()
+ b := DataStore{Engine: eng, EditDuration: 10 * time.Millisecond, ImageService: imgSvc}
+
+ // create comment with three images without preparing it properly
+ comment := store.Comment{
+ ID: "id-0",
+ Text: `
+ 
+ 
`,
+ Timestamp: time.Date(2017, 12, 20, 15, 18, 22, 0, time.Local),
+ Locator: store.Locator{URL: "https://radio-t.com", SiteID: "radio-t"},
+ User: store.User{ID: "user1", Name: "user name"},
+ }
+ _, err := b.Engine.Create(comment)
+ require.NoError(t, err)
+
+ // resubmit single comment with three images, of which two are in staging storage
+ mockStore.On("Info").Once().Return(image.StoreInfo{FirstStagingImageTS: time.Time{}.Add(time.Second)}, nil)
+ err = b.ResubmitStagingImages([]string{"radio-t"})
+ assert.NoError(t, err)
+
+ // wait for Submit goroutine to commit image
+ mockStore.On("Commit", "dev_user/bqf122eq9r8ad657n3ng").Once().Return(nil)
+ mockStore.On("Commit", "dev_user/bqf321eq9r8ad657n3ng").Once().Return(nil)
+ time.Sleep(time.Millisecond * 100)
+
+ mockStore.AssertNumberOfCalls(t, "Info", 1)
+ mockStore.AssertNumberOfCalls(t, "Commit", 2)
+
+ // empty answer
+ mockStoreEmpty := image.MockStore{}
+ imgSvcEmpty := image.NewService(&mockStoreEmpty,
+ image.ServiceParams{
+ EditDuration: 10 * time.Millisecond,
+ ImageAPI: "http://127.0.0.1:8080/api/v1/picture/",
+ })
+ defer imgSvcEmpty.Close(context.TODO())
+ bEmpty := DataStore{Engine: eng, EditDuration: 10 * time.Millisecond, ImageService: imgSvcEmpty}
+
+ // resubmit receive empty timestamp and should do nothing
+ mockStoreEmpty.On("Info").Once().Return(image.StoreInfo{FirstStagingImageTS: time.Time{}}, nil)
+ err = bEmpty.ResubmitStagingImages([]string{"radio-t", "non_existent"})
+ assert.NoError(t, err)
+
+ mockStoreEmpty.AssertNumberOfCalls(t, "Info", 1)
+
+ // error from image storage
+ mockStoreError := image.MockStore{}
+ imgSvcError := image.NewService(&mockStoreError,
+ image.ServiceParams{
+ EditDuration: 10 * time.Millisecond,
+ ImageAPI: "http://127.0.0.1:8080/api/v1/picture/",
+ })
+ defer imgSvcError.Close(context.TODO())
+ bError := DataStore{Engine: eng, EditDuration: 10 * time.Millisecond, ImageService: imgSvcError}
+
+ // resubmit will receive error from image storage and should return it
+ mockStoreError.On("Info").Once().Return(image.StoreInfo{}, errors.New("mock_err"))
+ err = bError.ResubmitStagingImages([]string{"radio-t"})
+ assert.EqualError(t, err, "mock_err")
+
+ mockStoreError.AssertNumberOfCalls(t, "Info", 1)
+}
+
+func TestService_ResubmitStagingImages_EngineError(t *testing.T) {
+ mockStore := image.MockStore{}
+ imgSvc := image.NewService(&mockStore,
+ image.ServiceParams{
+ EditDuration: 10 * time.Millisecond,
+ ImageAPI: "http://127.0.0.1:8080/api/v1/picture/",
+ })
+ defer imgSvc.Close(context.TODO())
+
+ engineMock := engine.MockInterface{}
+ site1Req := engine.FindRequest{Locator: store.Locator{SiteID: "site1", URL: ""}, Sort: "time", Since: time.Time{}.Add(time.Second)}
+ site2Req := engine.FindRequest{Locator: store.Locator{SiteID: "site2", URL: ""}, Sort: "time", Since: time.Time{}.Add(time.Second)}
+ engineMock.On("Find", site1Req).Return(nil, nil)
+ engineMock.On("Find", site2Req).Return(nil, errors.New("mockError"))
+ b := DataStore{Engine: &engineMock, EditDuration: 10 * time.Millisecond, ImageService: imgSvc}
+
+ // One call without error and one with error
+ mockStore.On("Info").Once().Return(image.StoreInfo{FirstStagingImageTS: time.Time{}.Add(time.Second)}, nil)
+ err := b.ResubmitStagingImages([]string{"site1", "site2"})
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), "problem finding comments for site site2: mockError")
+
+ mockStore.AssertNumberOfCalls(t, "Info", 1)
+}
+
func TestService_alterComment(t *testing.T) {
engineMock := engine.MockInterface{}