Merge pull request #693 from umputun/img_commit_on_start

Image commit on start

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
+}

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())
+}

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)
+}

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())
+}

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,
 	})
 }

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":

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

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,

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")
+}

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)

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.

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")

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 ")

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)

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)

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
+}

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)

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) {
 

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: `<img src="http://127.0.0.1:8080/api/v1/picture/dev_user/bqf122eq9r8ad657n3ng" alt="startrails_01.jpg"><br/>
+               <img src="http://127.0.0.1:8080/api/v1/picture/dev_user/bqf321eq9r8ad657n3ng" alt="cat.png"><br/>
+               <img src="https://homepages.cae.wisc.edu/~ece533/images/boat.png" alt="boat.png">`,
+		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{}