Impl feed job

Author: gedorinku

app/di/store_component.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/ProgrammingLab/prolab-accounts/app/config"
 	"github.com/ProgrammingLab/prolab-accounts/infra/store"
+	entrystore "github.com/ProgrammingLab/prolab-accounts/infra/store/entry"
 	feedstore "github.com/ProgrammingLab/prolab-accounts/infra/store/feed"
 	profilestore "github.com/ProgrammingLab/prolab-accounts/infra/store/profile"
 	sessionstore "github.com/ProgrammingLab/prolab-accounts/infra/store/session"
@@ -25,6 +26,7 @@ type StoreComponent interface {
 	ProfileStore(ctx context.Context) store.ProfileStore
 	UserBlogStore(ctx context.Context) store.UserBlogStore
 	FeedStore(ctx context.Context) store.FeedStore
+	EntryStore(ctx context.Context) store.EntryStore
 }
 
 // NewStoreComponent returns new store component
@@ -145,3 +147,7 @@ func (s *storeComponentImpl) UserBlogStore(ctx context.Context) store.UserBlogSt
 func (s *storeComponentImpl) FeedStore(ctx context.Context) store.FeedStore {
 	return feedstore.NewFeedStore(ctx)
 }
+
+func (s *storeComponentImpl) EntryStore(ctx context.Context) store.EntryStore {
+	return entrystore.NewEntryStore(ctx, s.db)
+}

app/job/feed_job.go

@@ -0,0 +1,38 @@
+package job
+
+import (
+	"context"
+
+	"google.golang.org/grpc/grpclog"
+
+	"github.com/ProgrammingLab/prolab-accounts/app/di"
+)
+
+func feedJob(ctx context.Context, store di.StoreComponent) error {
+	bs := store.UserBlogStore(ctx)
+	blogs, err := bs.ListUserBlogs()
+	if err != nil {
+		return err
+	}
+
+	fs := store.FeedStore(ctx)
+	es := store.EntryStore(ctx)
+	for _, b := range blogs {
+		if b.FeedURL == "" {
+			continue
+		}
+		feed, err := fs.GetFeed(b.FeedURL)
+		if err != nil {
+			grpclog.Errorf("feed job: failed to get feed: blog id: %v : %+v", b.ID, err)
+			continue
+		}
+
+		n, err := es.CreateEntries(b, feed)
+		if err != nil {
+			return err
+		}
+		grpclog.Infof("feed job: created %v entries", n)
+	}
+
+	return nil
+}

app/job/job.go

@@ -1,6 +1,7 @@
 package job
 
 import (
+	"context"
 	"time"
 
 	"google.golang.org/grpc/grpclog"
@@ -13,11 +14,13 @@ import (
 var (
 	started = false
 	stop    = make(chan struct{})
-	jobs    = []Job{}
+	jobs    = []Job{
+		feedJob,
+	}
 )
 
 // Job represents job for worker
-type Job func(store di.StoreComponent) error
+type Job func(ctx context.Context, store di.StoreComponent) error
 
 // Start starts the worker
 func Start(store di.StoreComponent, cfg *config.Config) {
@@ -52,7 +55,7 @@ func run(store di.StoreComponent, interval time.Duration) {
 		select {
 		case <-time.After(interval):
 			for _, j := range jobs {
-				err := j(store)
+				err := j(context.Background(), store)
 				if err != nil {
 					grpclog.Errorf("job error: %+v", err)
 				}

db/blogs.schema

@@ -3,6 +3,8 @@ create_table :blogs, force: :cascade do |t|
   t.string :feed_url, null: false
   t.references :user, foreign_key: true, null: false
   t.timestamps
+
+  t.index [:feed_url], unique: true
 end
 
 add_foreign_key :blogs, :users

db/entries.schema

@@ -7,6 +7,7 @@ create_table :entries, force: :cascade do |t|
   t.string :guid, null: false
   t.string :image_url, null: false
   t.references :blog, foreign_key: true, null: false
+  t.datetime :published_at, null: true
   t.timestamps
 
   t.index [:guid], unique: true

infra/record/entries.go

@@ -0,0 +1,116 @@
+package entrystore
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/base64"
+	"encoding/binary"
+
+	"github.com/mmcdole/gofeed"
+	"github.com/pkg/errors"
+	"github.com/volatiletech/null"
+	"github.com/volatiletech/sqlboiler/boil"
+	"github.com/volatiletech/sqlboiler/queries/qm"
+
+	"github.com/ProgrammingLab/prolab-accounts/app/util"
+	"github.com/ProgrammingLab/prolab-accounts/infra/record"
+	"github.com/ProgrammingLab/prolab-accounts/infra/store"
+)
+
+type entryStoreImpl struct {
+	ctx context.Context
+	db  *sql.DB
+}
+
+// NewEntryStore returns new entry blog store
+func NewEntryStore(ctx context.Context, db *sql.DB) store.EntryStore {
+	return &entryStoreImpl{
+		ctx: ctx,
+		db:  db,
+	}
+}
+
+func (s *entryStoreImpl) CreateEntries(blog *record.Blog, feed *gofeed.Feed) (n int64, err error) {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return 0, errors.WithStack(err)
+	}
+	defer func() {
+		if e := util.ErrorFromRecover(recover()); e != nil {
+			_ = tx.Rollback()
+			err = e
+		}
+	}()
+
+	mods := []qm.QueryMod{
+		qm.Select(record.EntryColumns.ID, record.EntryColumns.GUID),
+		qm.Where("blog_id = ?", blog.ID),
+	}
+	entries, err := record.Entries(mods...).All(s.ctx, tx)
+	if err != nil {
+		_ = tx.Rollback()
+		return 0, errors.WithStack(err)
+	}
+
+	exists := make(map[string]struct{})
+	for _, e := range entries {
+		exists[e.GUID] = struct{}{}
+	}
+
+	n = 0
+	for _, item := range feed.Items {
+		guid, err := getGUID(blog.ID, item.GUID)
+		if err != nil {
+			_ = tx.Rollback()
+			return 0, err
+		}
+
+		_, ok := exists[guid]
+		if ok {
+			continue
+		}
+
+		e := &record.Entry{
+			Title:       item.Title,
+			Description: item.Description,
+			Content:     item.Content,
+			Link:        item.Link,
+			AuthorID:    blog.UserID,
+			GUID:        guid,
+			BlogID:      blog.ID,
+			PublishedAt: null.TimeFromPtr(item.PublishedParsed),
+		}
+		if i := item.Image; i != nil {
+			e.ImageURL = i.URL
+		}
+
+		err = e.Insert(s.ctx, tx, boil.Infer())
+		if err != nil {
+			_ = tx.Rollback()
+			return 0, errors.WithStack(err)
+		}
+		n++
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		_ = tx.Rollback()
+		return 0, errors.WithStack(err)
+	}
+	return n, nil
+}
+
+func getGUID(blogID int64, guid string) (string, error) {
+	h := sha256.New()
+	err := binary.Write(h, binary.LittleEndian, blogID)
+	if err != nil {
+		return "", errors.WithStack(err)
+	}
+	_, err = h.Write([]byte(guid))
+	if err != nil {
+		return "", errors.WithStack(err)
+	}
+
+	return base64.RawURLEncoding.EncodeToString(h.Sum(nil)), nil
+}

infra/record/entries_test.go

@@ -0,0 +1,11 @@
+package store
+
+import (
+	"github.com/ProgrammingLab/prolab-accounts/infra/record"
+	"github.com/mmcdole/gofeed"
+)
+
+// EntryStore accesses entry data
+type EntryStore interface {
+	CreateEntries(blog *record.Blog, feed *gofeed.Feed) (int64, error)
+}

infra/store/entry/entry_store.go

@@ -26,6 +26,14 @@ func NewUserBlogStore(ctx context.Context, db *sql.DB) store.UserBlogStore {
 	}
 }
 
+func (s *userBlogStoreImpl) ListUserBlogs() ([]*record.Blog, error) {
+	b, err := record.Blogs().All(s.ctx, s.db)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return b, nil
+}
+
 func (s *userBlogStoreImpl) GetUserBlog(blogID int64) (*record.Blog, error) {
 	b, err := record.FindBlog(s.ctx, s.db, int64(blogID))
 	if err != nil {
@@ -44,14 +52,15 @@ func (s *userBlogStoreImpl) CreateUserBlog(blog *record.Blog) error {
 	return nil
 }
 
-func (s *userBlogStoreImpl) UpdateUserBlog(blog *record.Blog) error {
+func (s *userBlogStoreImpl) UpdateUserBlog(blog *record.Blog) (err error) {
 	tx, err := s.db.Begin()
 	if err != nil {
 		return errors.WithStack(err)
 	}
 	defer func() {
-		if err = util.ErrorFromRecover(recover()); err != nil {
+		if e := util.ErrorFromRecover(recover()); e != nil {
 			_ = tx.Rollback()
+			err = e
 		}
 	}()
 

infra/store/entry_store.go

@@ -6,6 +6,7 @@ import (
 
 // UserBlogStore accesses users data
 type UserBlogStore interface {
+	ListUserBlogs() ([]*record.Blog, error)
 	GetUserBlog(blogID int64) (*record.Blog, error)
 	CreateUserBlog(blog *record.Blog) error
 	UpdateUserBlog(blog *record.Blog) error