From 9b9479367ad1b217efe15b1c69f6e80e2098f096 Mon Sep 17 00:00:00 2001 From: Umputun Date: Mon, 30 Apr 2018 00:54:15 -0500 Subject: [PATCH] in-proc import with initial delete --- app/main.go | 2 ++ app/migrator/disqus.go | 5 ++++- app/migrator/migrator.go | 1 + app/migrator/remark.go | 5 +++++ app/rest/api/admin.go | 22 ++++++++++++++-------- app/rest/api/rest.go | 14 +++++++++++--- app/store/bolt.go | 26 +++++++++++++++++++++++++- app/store/bolt_test.go | 21 +++++++++++++++++++++ app/store/store.go | 1 + 9 files changed, 84 insertions(+), 13 deletions(-) diff --git a/app/main.go b/app/main.go index 333b7e38..e3a87156 100644 --- a/app/main.go +++ b/app/main.go @@ -120,6 +120,8 @@ func main() { Version: revision, DataService: dataService, Exporter: exporter, + Importer: &migrator.Remark{CommentCreator: dataStore}, + Disqus: &migrator.Disqus{CommentCreator: dataStore}, WebRoot: opts.ServerCommand.WebRoot, Authenticator: auth.Authenticator{ Admins: opts.Admins, diff --git a/app/migrator/disqus.go b/app/migrator/disqus.go index 736b6e4e..fd840dd0 100644 --- a/app/migrator/disqus.go +++ b/app/migrator/disqus.go @@ -53,6 +53,10 @@ type uid struct { // Import from disqus and save to store func (d *Disqus) Import(r io.Reader, siteID string) (err error) { + if err = d.DeleteAll(siteID); err != nil { + return err + } + commentsCh := d.convert(r, siteID) failed, passed := 0, 0 for c := range commentsCh { @@ -145,7 +149,6 @@ func (d *Disqus) convert(r io.Reader, siteID string) (ch chan store.Comment) { log.Printf("[DEBUG] processed %d comments", stats.commentsCount) } } - } } close(commentsCh) diff --git a/app/migrator/migrator.go b/app/migrator/migrator.go index ee59b8b9..41bf93aa 100644 --- a/app/migrator/migrator.go +++ b/app/migrator/migrator.go @@ -26,6 +26,7 @@ type Exporter interface { // CommentCreator is a minimal interface used by importer to make comments type CommentCreator interface { Create(comment store.Comment) (commentID string, err error) + DeleteAll(siteID string) error } // CommentFinder is a minimal interface used by exporter to find comments and list posts diff --git a/app/migrator/remark.go b/app/migrator/remark.go index 3923368b..f92aca04 100644 --- a/app/migrator/remark.go +++ b/app/migrator/remark.go @@ -56,6 +56,11 @@ func (r *Remark) Export(w io.Writer, siteID string) error { // Import comments from json strings produced by Remark.Export func (r *Remark) Import(reader io.Reader, siteID string) error { + + if err := r.DeleteAll(siteID); err != nil { + return err + } + failed := 0 total, comments := 0, 0 scanner := bufio.NewScanner(reader) diff --git a/app/rest/api/admin.go b/app/rest/api/admin.go index 803a6f6d..8eada1e8 100644 --- a/app/rest/api/admin.go +++ b/app/rest/api/admin.go @@ -18,11 +18,12 @@ import ( // admin provides router for all requests available for admin users only type admin struct { - dataService store.Service - exporter migrator.Exporter - importer migrator.Importer - cache rest.LoadingCache - defAvatarURL string + dataService store.Service + exporterNative migrator.Exporter + importerNative migrator.Importer + importerDisqus migrator.Importer + cache rest.LoadingCache + defAvatarURL string } func (a *admin) routes(middlewares ...func(http.Handler) http.Handler) chi.Router { @@ -107,16 +108,21 @@ func (a *admin) exportCtrl(w http.ResponseWriter, r *http.Request) { writer = gzip.NewWriter(w) } - if err := a.exporter.Export(writer, siteID); err != nil { + if err := a.exporterNative.Export(writer, siteID); err != nil { rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "export failed") } } -// POST /import?site=site-id +// POST /import?site=site-id&provider=disqus|remark // imports comments from post body. func (a *admin) importCtrl(w http.ResponseWriter, r *http.Request) { siteID := r.URL.Query().Get("site") - if err := a.importer.Import(r.Body, siteID); err != nil { + importer := a.importerNative + if r.URL.Query().Get("provider") == "disqus" { + importer = a.importerDisqus + } + + if err := importer.Import(r.Body, siteID); err != nil { rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "import failed") } a.cache.Flush() diff --git a/app/rest/api/rest.go b/app/rest/api/rest.go index 2036797c..bc7822d3 100644 --- a/app/rest/api/rest.go +++ b/app/rest/api/rest.go @@ -32,10 +32,13 @@ type Rest struct { Version string DataService store.Service Authenticator auth.Authenticator - Exporter migrator.Exporter Cache rest.LoadingCache WebRoot string + Exporter migrator.Exporter + Importer migrator.Importer + Disqus migrator.Importer + httpServer *http.Server amdminService admin } @@ -48,8 +51,13 @@ func (s *Rest) Run(port int) { log.Printf("[DEBUG] admins %+v", s.Authenticator.Admins) } - s.amdminService = admin{dataService: s.DataService, exporter: s.Exporter, cache: s.Cache, - defAvatarURL: s.Authenticator.AvatarProxy.Default(), + s.amdminService = admin{ + dataService: s.DataService, + exporterNative: s.Exporter, + importerNative: s.Importer, + importerDisqus: s.Disqus, + cache: s.Cache, + defAvatarURL: s.Authenticator.AvatarProxy.Default(), } router := chi.NewRouter() diff --git a/app/store/bolt.go b/app/store/bolt.go index 1904d181..fa9cc300 100644 --- a/app/store/bolt.go +++ b/app/store/bolt.go @@ -38,6 +38,8 @@ const ( userLimit = 50 ) +var topBuckets = []string{postsBucketName, lastBucketName, userBucketName, blocksBucketName, countsBucketName} + const tsNano = "2006-01-02T15:04:05.000000000Z07:00" // BoltSite defines single site param @@ -59,7 +61,6 @@ func NewBoltDB(options bolt.Options, sites ...BoltSite) (*BoltDB, error) { // make top-level buckets err = db.Update(func(tx *bolt.Tx) error { - topBuckets := []string{postsBucketName, lastBucketName, userBucketName, blocksBucketName, countsBucketName} for _, bktName := range topBuckets { if _, e := tx.CreateBucketIfNotExists([]byte(bktName)); e != nil { return errors.Wrapf(err, "failed to create top level bucket %s", bktName) @@ -175,6 +176,29 @@ func (b *BoltDB) Delete(locator Locator, commentID string) error { }) } +// DeleteAll removes all top-level buckets for given siteID +func (b *BoltDB) DeleteAll(siteID string) error { + + bdb, err := b.db(siteID) + if err != nil { + return err + } + // delete top-level buckets + err = bdb.Update(func(tx *bolt.Tx) error { + for _, bktName := range topBuckets { + if e := tx.DeleteBucket([]byte(bktName)); e != nil { + return errors.Wrapf(err, "failed to delete top level bucket %s", bktName) + } + if _, e := tx.CreateBucketIfNotExists([]byte(bktName)); e != nil { + return errors.Wrapf(err, "failed to create top level bucket %s", bktName) + } + } + return nil + }) + + return errors.Wrapf(err, "failed to delete top level buckets fro site %s", siteID) +} + // Find returns all comments for post and sorts results func (b *BoltDB) Find(locator Locator, sortFld string) (comments []Comment, err error) { comments = []Comment{} diff --git a/app/store/bolt_test.go b/app/store/bolt_test.go index cb24de5a..7896f02a 100644 --- a/app/store/bolt_test.go +++ b/app/store/bolt_test.go @@ -52,6 +52,27 @@ func TestBoltDB_Delete(t *testing.T) { assert.Equal(t, 1, len(comments), "1 in last, 1 removed") } +func TestBoltDB_DeleteAll(t *testing.T) { + defer os.Remove(testDb) + b := prep(t) + + loc := Locator{URL: "https://radio-t.com", SiteID: "radio-t"} + res, err := b.Find(loc, "time") + assert.Nil(t, err) + assert.Equal(t, 2, len(res), "initially 2 comments") + + err = b.DeleteAll("radio-t") + assert.Nil(t, err) + + comments, err := b.Last("radio-t", 10) + assert.Nil(t, err) + assert.Equal(t, 0, len(comments), "nothing left") + + c, err := b.Count(Locator{URL: "https://radio-t.com", SiteID: "radio-t"}) + assert.Nil(t, err) + assert.Equal(t, 0, c, "0 count") +} + func TestBoltDB_Get(t *testing.T) { defer os.Remove(testDb) b := prep(t) diff --git a/app/store/store.go b/app/store/store.go index bdb9a89f..e6e961d1 100644 --- a/app/store/store.go +++ b/app/store/store.go @@ -28,6 +28,7 @@ type Accessor interface { // Admin defines all store ops avail for admin only type Admin interface { Delete(locator Locator, commentID string) error // delete comment by id + DeleteAll(siteID string) error // delete all data from site SetBlock(siteID string, userID string, status bool) error // block or unblock user IsBlocked(siteID string, userID string) bool // check if user blocked Blocked(siteID string) ([]BlockedUser, error) // get list of blocked users