use worker

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

internal/config/config.go

@@ -1,8 +1,14 @@
 package config
 
-import "time"
+import (
+	"time"
 
-type AppConfig struct{}
+	mtime "go.unistack.org/micro/v4/util/time"
+)
+
+type AppConfig struct {
+	CheckInterval mtime.Duration `json:"check_interval" yaml:"check_interval" default:"1d"`
+}
 
 type ServerConfig struct {
 	Name        string `json:"name" yaml:"name"`
@@ -24,7 +30,7 @@ type TracerConfig struct {
 type VaultConfig struct {
 	Addr  string `env:"VAULT_ADDR" json:"addr" yaml:"addr" default:"http://127.0.0.1:8200"`
 	Token string `env:"VAULT_TOKEN" json:"-" yaml:"-"`
-	Path  string `env:"VAULT_PATH" json:"-" yaml:"-" default:"apigw/data/authn"`
+	Path  string `env:"VAULT_PATH" json:"-" yaml:"-" default:"pkgdash/data/pkgdash"`
 }
 
 type MeterConfig struct {

internal/database/database.go

@@ -0,0 +1,253 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+
+	appconfig "git.unistack.org/unistack-org/pkgdash/internal/config"
+	"github.com/golang-migrate/migrate/v4"
+	"github.com/golang-migrate/migrate/v4/database"
+	mpgx "github.com/golang-migrate/migrate/v4/database/pgx"
+	msqlite "github.com/golang-migrate/migrate/v4/database/sqlite"
+	"github.com/golang-migrate/migrate/v4/source/iofs"
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/stdlib"
+	"github.com/jmoiron/sqlx"
+	"go.unistack.org/micro/v4/logger"
+	_ "modernc.org/sqlite"
+)
+
+func ParseDSN(cfg *appconfig.DatabaseConfig) error {
+	var err error
+
+	u, err := url.Parse(cfg.DSN)
+	if err != nil {
+		return err
+	}
+
+	values := u.Query()
+	var value string
+
+	if value = values.Get("conn_max"); value != "" {
+		values.Del("conn_max")
+		maxOpenConns, err := strconv.Atoi(value)
+		if err != nil {
+			return err
+		}
+		cfg.MaxOpenConns = maxOpenConns
+		cfg.MaxIdleConns = maxOpenConns / 2
+	}
+
+	if value = values.Get("conn_maxidle"); value != "" {
+		values.Del("conn_maxidle")
+		maxIdleConns, err := strconv.Atoi(value)
+		if err != nil {
+			return err
+		}
+		cfg.MaxIdleConns = maxIdleConns
+	}
+
+	if value = values.Get("conn_lifetime"); value != "" {
+		values.Del("conn_lifetime")
+		connMaxLifetime, err := time.ParseDuration(value)
+		if err != nil {
+			return err
+		}
+		cfg.ConnMaxLifetime = connMaxLifetime
+	}
+
+	if value = values.Get("conn_maxidletime"); value != "" {
+		values.Del("conn_maxidletime")
+		connMaxIdleTime, err := time.ParseDuration(value)
+		if err != nil {
+			return err
+		}
+		cfg.ConnMaxIdleTime = connMaxIdleTime
+	}
+
+	if mtype := values.Get("migrate"); mtype != "" {
+		values.Del("migrate")
+		cfg.Migrate = mtype
+	}
+
+	switch u.Scheme {
+	case "postgres", "pgsql", "postgresql":
+		u.Scheme = "postgres"
+	case "sqlite", "sqlite3":
+		u.Scheme = "sqlite"
+	default:
+		return fmt.Errorf("unknown database %s", u.Scheme)
+	}
+
+	cfg.Type = u.Scheme
+	u.RawQuery = values.Encode()
+
+	cfg.ConnStr = u.String()
+
+	return nil
+}
+
+func connect(ctx context.Context, cfg *appconfig.DatabaseConfig, log logger.Logger) (*sqlx.DB, error) {
+	var db *sqlx.DB
+	var err error
+
+	log.Infof(ctx, "connect to %s", cfg.Type)
+	switch cfg.Type {
+	case "postgres", "pgsql", "postgresql":
+		db, err = connectPostgres(ctx, cfg.ConnStr)
+		cfg.Type = "postgres"
+	case "sqlite", "sqlite3":
+		db, err = connectSqlite(ctx, cfg.ConnStr)
+		cfg.Type = "sqlite"
+	default:
+		return nil, fmt.Errorf("unknown database type %s", cfg.Type)
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}
+
+func Connect(ctx context.Context, cfg *appconfig.DatabaseConfig, log logger.Logger) (*sqlx.DB, error) {
+	db, err := connect(ctx, cfg, log)
+	if err != nil {
+		return nil, err
+	}
+	m, err := migratePrepare(ctx, db, log, cfg.Type)
+	if err != nil {
+		return nil, err
+	}
+
+	switch cfg.Migrate {
+	case "":
+		break
+	case "up":
+		logger.Infof(ctx, "migrate up")
+		err = m.Up()
+	case "down":
+		logger.Infof(ctx, "migrate down")
+		err = m.Down()
+	case "seed":
+		logger.Infof(ctx, "migrate seed")
+		if err = m.Drop(); err == nil {
+			err = m.Up()
+		}
+	default:
+		logger.Infof(ctx, "migrate version")
+		v, verr := strconv.ParseUint(cfg.Type, 10, 64)
+		if verr != nil {
+			return nil, err
+		}
+		err = m.Migrate(uint(v))
+	}
+
+	if err == nil || err == migrate.ErrNoChange {
+		srcerr, dberr := m.Close()
+		if srcerr != nil {
+			err = srcerr
+		} else if dberr != nil {
+			err = dberr
+		} else {
+			err = nil
+		}
+	}
+
+	if err == nil {
+		db, err = connect(ctx, cfg, log)
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	db.SetConnMaxIdleTime(cfg.ConnMaxIdleTime)
+	db.SetConnMaxLifetime(cfg.ConnMaxLifetime)
+	db.SetMaxIdleConns(cfg.MaxIdleConns)
+	db.SetMaxOpenConns(cfg.MaxOpenConns)
+
+	return db, nil
+}
+
+func connectSqlite(ctx context.Context, connstr string) (*sqlx.DB, error) {
+	if !strings.Contains(connstr, ":memory:") {
+		return sqlx.ConnectContext(ctx, "sqlite", "file:"+connstr[9:])
+	}
+	return sqlx.ConnectContext(ctx, "sqlite", connstr[9:])
+}
+
+func connectPostgres(ctx context.Context, connstr string) (*sqlx.DB, error) {
+	// parse connection string
+	dbConf, err := pgx.ParseConfig(connstr)
+	if err != nil {
+		return nil, err
+	}
+
+	// needed for pgbouncer
+	dbConf.RuntimeParams = map[string]string{
+		"standard_conforming_strings": "on",
+		"application_name":            "authn",
+	}
+
+	// may be needed for pbbouncer, needs to check
+	// dbConf.PreferSimpleProtocol = true
+	// register pgx conn
+	connStr := stdlib.RegisterConnConfig(dbConf)
+
+	db, err := sqlx.ConnectContext(ctx, "pgx", connStr)
+	if err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}
+
+func migratePrepare(ctx context.Context, db *sqlx.DB, log logger.Logger, dbtype string) (*migrate.Migrate, error) {
+	var driver database.Driver
+	var err error
+
+	switch dbtype {
+	case "postgres":
+		driver, err = mpgx.WithInstance(db.DB, &mpgx.Config{
+			DatabaseName:    "pkgdash",
+			MigrationsTable: "schema_migrations",
+		})
+	case "sqlite":
+		driver, err = msqlite.WithInstance(db.DB, &msqlite.Config{
+			DatabaseName:    "pkgdash",
+			MigrationsTable: "schema_migrations",
+		})
+	}
+
+	if err != nil {
+		return nil, err
+	}
+	source, err := iofs.New(assets, "migrations/"+dbtype)
+	if err != nil {
+		return nil, err
+	}
+	m, err := migrate.NewWithInstance("fs", source, "apigw", driver)
+	if err != nil {
+		return nil, err
+	}
+	m.Log = &mLog{ctx: ctx, l: log}
+	return m, nil
+}
+
+type mLog struct {
+	ctx context.Context
+	l   logger.Logger
+}
+
+func (l *mLog) Verbose() bool {
+	return l.l.V(logger.DebugLevel)
+}
+
+func (l *mLog) Printf(format string, v ...interface{}) {
+	l.l.Infof(l.ctx, format, v...)
+}

internal/database/embed.go

@@ -0,0 +1,8 @@
+package database
+
+import (
+	"embed"
+)
+
+//go:embed migrations
+var assets embed.FS

internal/database/migrations/sqlite/000001_init_schema.down.sql

@@ -0,0 +1,4 @@
+drop table if exists packages;
+drop table if exists modules;
+drop table if exists issues;
+drop table if exists comments;

internal/database/migrations/sqlite/000001_init_schema.up.sql

@@ -0,0 +1,38 @@
+create table if not exists comments (
+    id integer primary key autoincrement not null,
+    comment text,
+    package integer not null,
+    created timestamp not null default current_timestamp,
+    updated timestamp not null default current_timestamp
+);
+
+create table if not exists modules (
+    id integer primary key autoincrement not null,
+    name varchar not null ,
+    version varchar not null,
+    package integer not null,
+    last_version varchar not null,
+    created timestamp not null default current_timestamp,
+    updated timestamp not null default current_timestamp
+);
+
+create table if not exists issues (
+    id integer primary key autoincrement not null,
+    status integer default 0,
+    comment varchar,
+    created timestamp not null default current_timestamp,
+    updated timestamp not null default current_timestamp
+);
+
+create table if not exists packages (
+    id integer primary key autoincrement not null,
+    name varchar not null,
+    url varchar not null,
+    modules integer default 0,
+    issues integer default 0,
+    comments integer default 0,
+    created timestamp not null default current_timestamp,
+    updated timestamp not null default current_timestamp,
+    status integer default 1,
+    last_check timestamp
+);

internal/handler/comments_create.go

@@ -6,13 +6,13 @@ import (
 	"errors"
 	"net/http"
 
-	httpsrv "go.unistack.org/micro-server-http/v4"
 	"git.unistack.org/unistack-org/pkgdash/internal/models"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
 )
 
 func (h *Handler) CommentsCreate(ctx context.Context, req *pb.CommentsCreateReq, rsp *pb.CommentsCreateRsp) error {
-	logger := h.svc.Logger()
 	logger.Debug(ctx, "Start AddComment")
 
 	err := req.Validate()

internal/handler/comments_delete.go

@@ -6,12 +6,12 @@ import (
 	"errors"
 	"net/http"
 
-	httpsrv "go.unistack.org/micro-server-http/v4"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
 )
 
 func (h *Handler) CommentsDelete(ctx context.Context, req *pb.CommentsDeleteReq, rsp *pb.CommentsDeleteRsp) error {
-	logger := h.svc.Logger()
 	logger.Debug(ctx, "Start AddComment")
 
 	err := req.Validate()

internal/handler/comments_list.go

@@ -4,13 +4,13 @@ import (
 	"context"
 	"net/http"
 
-	httpsrv "go.unistack.org/micro-server-http/v4"
 	"git.unistack.org/unistack-org/pkgdash/internal/models"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
 )
 
 func (h *Handler) CommentsList(ctx context.Context, req *pb.CommentsListReq, rsp *pb.CommentsListRsp) error {
-	logger := h.svc.Logger()
 	logger.Debug(ctx, "Start GetModule")
 
 	err := req.Validate()

internal/handler/handler.go

@@ -1,27 +1,19 @@
 package handler
 
 import (
-	"context"
-	"errors"
 	"net/http"
 	"strconv"
 
-	"github.com/google/uuid"
-	"go.unistack.org/micro/v4"
-	cligit "git.unistack.org/unistack-org/pkgdash/internal/service/client_git"
 	"git.unistack.org/unistack-org/pkgdash/internal/storage"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
-	"google.golang.org/protobuf/encoding/protojson"
+	"github.com/google/uuid"
+	jsonpbcodec "go.unistack.org/micro-codec-jsonpb/v4"
+	"go.unistack.org/micro/v4/codec"
 )
 
 type Handler struct {
-	svc   micro.Service
-	store storage.Storage
-
-	protojson.MarshalOptions
-	protojson.UnmarshalOptions
-
-	git     cligit.Client
+	store   storage.Storage
+	codec   codec.Codec
 	chanUrl chan *pb.PackagesCreateReq
 }
 
@@ -52,25 +44,11 @@ func NewValidationError(err error) *pb.ErrorRsp {
 	}
 }
 
-func NewHandler(svc micro.Service, client cligit.Client) *Handler {
+func NewHandler(store storage.Storage) (*Handler, error) {
 	h := &Handler{
-		svc: svc,
-		git: client,
-	}
-	h.EmitUnpopulated = true
-	h.UseProtoNames = false
-	return h
-}
-
-func (h *Handler) Init(ctx context.Context) error {
-	store, err := storage.FromContext(h.svc.Options().Context)
-	if err != nil {
-		return errors.New("missing storage")
+		codec: jsonpbcodec.NewCodec(),
+		store: store,
 	}
 
-	h.chanUrl = h.git.Run(ctx, store)
-
-	h.store = store
-
-	return nil
+	return h, nil
 }

internal/handler/modules_list.go

@@ -4,13 +4,13 @@ import (
 	"context"
 	"net/http"
 
-	httpsrv "go.unistack.org/micro-server-http/v4"
 	"git.unistack.org/unistack-org/pkgdash/internal/models"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
 )
 
 func (h *Handler) ModulesList(ctx context.Context, req *pb.ModulesListReq, rsp *pb.ModulesListRsp) error {
-	logger := h.svc.Logger()
 	logger.Debug(ctx, "Start GetModule")
 
 	err := req.Validate()

internal/handler/packages_create.go

@@ -4,29 +4,30 @@ import (
 	"context"
 	"net/http"
 
-	httpsrv "go.unistack.org/micro-server-http/v4"
+	"git.unistack.org/unistack-org/pkgdash/internal/models"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
 )
 
 func (h *Handler) PackagesCreate(ctx context.Context, req *pb.PackagesCreateReq, rsp *pb.PackagesCreateRsp) error {
-	logger := h.svc.Logger()
-	logger.Debug(ctx, "Start AddPackage")
+	logger.Debug(ctx, "PackagesCreate handler start")
 
-	err := req.Validate()
+	if err := req.Validate(); err != nil {
+		logger.Error(ctx, err)
+		httpsrv.SetRspCode(ctx, http.StatusBadRequest)
+		return httpsrv.SetError(NewValidationError(err))
+	}
+
+	pkg, err := h.store.PackagesCreate(ctx, req)
 	if err != nil {
 		logger.Error(ctx, err)
 		httpsrv.SetRspCode(ctx, http.StatusBadRequest)
 		return httpsrv.SetError(NewValidationError(err))
 	}
 
-	if h.git.IsClose() {
-		logger.Error(ctx, "chan is closed")
-	} else {
-		h.chanUrl <- req
-	}
+	rsp.Package = models.NewPackage(pkg)
 
-	rsp.Status = "Sent"
-
-	logger.Debug(ctx, "Success finish addPackage")
+	logger.Debug(ctx, "PackagesCreate handler stop")
 	return nil
 }

internal/handler/packages_delete.go

@@ -4,12 +4,12 @@ import (
 	"context"
 	"net/http"
 
-	httpsrv "go.unistack.org/micro-server-http/v4"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
 )
 
 func (h *Handler) PackagesDelete(ctx context.Context, req *pb.PackagesDeleteReq, rsp *pb.PackagesDeleteRsp) error {
-	logger := h.svc.Logger()
 	logger.Debug(ctx, "Start UpdatePackage")
 
 	if err := req.Validate(); err != nil {

internal/handler/packages_list.go

@@ -4,14 +4,14 @@ import (
 	"context"
 	"net/http"
 
-	httpsrv "go.unistack.org/micro-server-http/v4"
 	"git.unistack.org/unistack-org/pkgdash/internal/models"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
 )
 
 func (h *Handler) PackagesList(ctx context.Context, req *pb.PackagesListReq, rsp *pb.PackagesListRsp) error {
-	logger := h.svc.Logger()
-	logger.Debug(ctx, "Start getListPackage")
+	logger.Debug(ctx, "PackagesList handler start")
 
 	packages, err := h.store.PackagesList(ctx, req)
 	if err != nil {
@@ -23,6 +23,6 @@ func (h *Handler) PackagesList(ctx context.Context, req *pb.PackagesListReq, rsp
 	for _, pkg := range packages {
 		rsp.Packages = append(rsp.Packages, models.NewPackage(pkg))
 	}
-	logger.Debug(ctx, "Success finish getListPackage")
+	logger.Debug(ctx, "PackagesList handler stop")
 	return nil
 }

internal/handler/packages_lookup.go

@@ -0,0 +1,33 @@
+package handler
+
+import (
+	"context"
+	"net/http"
+
+	"git.unistack.org/unistack-org/pkgdash/internal/models"
+	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
+)
+
+func (h *Handler) PackagesLookup(ctx context.Context, req *pb.PackagesLookupReq, rsp *pb.PackagesLookupRsp) error {
+	logger.Debug(ctx, "Start PackagesLookup")
+
+	if err := req.Validate(); err != nil {
+		logger.Error(ctx, err)
+		httpsrv.SetRspCode(ctx, http.StatusBadRequest)
+		return httpsrv.SetError(NewValidationError(err))
+	}
+
+	pkg, err := h.store.PackagesLookup(ctx, req)
+	if err != nil {
+		logger.Error(ctx, err)
+		httpsrv.SetRspCode(ctx, http.StatusInternalServerError)
+		return httpsrv.SetError(NewInternalError(err))
+	}
+
+	rsp.Package = models.NewPackage(pkg)
+
+	logger.Debug(ctx, "Success finish PackagesLookup")
+	return nil
+}

internal/handler/packages_update.go

@@ -4,12 +4,13 @@ import (
 	"context"
 	"net/http"
 
-	httpsrv "go.unistack.org/micro-server-http/v4"
+	"git.unistack.org/unistack-org/pkgdash/internal/models"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	httpsrv "go.unistack.org/micro-server-http/v4"
+	"go.unistack.org/micro/v4/logger"
 )
 
 func (h *Handler) PackagesUpdate(ctx context.Context, req *pb.PackagesUpdateReq, rsp *pb.PackagesUpdateRsp) error {
-	logger := h.svc.Logger()
 	logger.Debug(ctx, "Start UpdatePackage")
 
 	if err := req.Validate(); err != nil {
@@ -18,13 +19,14 @@ func (h *Handler) PackagesUpdate(ctx context.Context, req *pb.PackagesUpdateReq,
 		return httpsrv.SetError(NewValidationError(err))
 	}
 
-	if err := h.store.PackagesUpdate(ctx, req); err != nil {
+	pkg, err := h.store.PackagesUpdate(ctx, req)
+	if err != nil {
 		logger.Error(ctx, err)
 		httpsrv.SetRspCode(ctx, http.StatusInternalServerError)
 		return httpsrv.SetError(NewInternalError(err))
 	}
 
-	// rsp.Id = req.Id
+	rsp.Package = models.NewPackage(pkg)
 
 	logger.Debug(ctx, "Success finish UpdatePackage")
 	return nil

internal/models/models.go

@@ -1,6 +1,7 @@
 package models
 
 import (
+	"database/sql"
 	"time"
 
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
@@ -8,18 +9,20 @@ import (
 )
 
 type Package struct {
-	Name     string    `db:"name" json:"name"`
-	URL      string    `db:"url" json:"url"`
-	Modules  []uint64  `db:"modules" json:"modules"`
-	Issues   []uint64  `db:"issues" json:"issues,omitempty"`
-	Comments []uint64  `db:"comments" json:"comments,omitempty"`
-	ID       uint64    `db:"id" json:"id"`
-	Created  time.Time `db:"created" json:"created"`
-	Updated  time.Time `db:"updated" json:"updated,omitempty"`
+	Created   time.Time    `db:"created"`
+	Updated   time.Time    `db:"updated"`
+	Name      string       `db:"name"`
+	URL       string       `db:"url"`
+	Modules   uint64       `db:"modules"`
+	Issues    uint64       `db:"issues"`
+	Comments  uint64       `db:"comments"`
+	ID        uint64       `db:"id"`
+	Status    uint64       `db:"status"`
+	LastCheck sql.NullTime `db:"last_check"`
 }
 
 func NewPackage(pkg *Package) *pb.Package {
-	return &pb.Package{
+	rsp := &pb.Package{
 		Name:     pkg.Name,
 		Url:      pkg.URL,
 		Modules:  pkg.Modules,
@@ -29,16 +32,20 @@ func NewPackage(pkg *Package) *pb.Package {
 		Created:  timestamppb.New(pkg.Created),
 		Updated:  timestamppb.New(pkg.Updated),
 	}
+	if pkg.LastCheck.Valid {
+		rsp.LastCheck = timestamppb.New(pkg.LastCheck.Time)
+	}
+	return rsp
 }
 
 type Module struct {
+	Created     time.Time `db:"created"`
+	Updated     time.Time `db:"updated"`
 	Name        string    `db:"name"`
 	Version     string    `db:"version"`
 	LastVersion string    `db:"last_version"`
 	ID          uint64    `db:"id"`
 	Package     uint64    `db:"package"`
-	Created     time.Time `db:"created" json:"created"`
-	Updated     time.Time `db:"updated" json:"updated,omitempty"`
 }
 
 func NewModule(mod *Module) *pb.Module {
@@ -54,24 +61,24 @@ func NewModule(mod *Module) *pb.Module {
 }
 
 type Issue struct {
-	Desc    string  `db:"desc"`
+	Comment string  `db:"comment"`
 	Modules []int64 `db:"modules"`
 	ID      uint64  `db:"id"`
 	Status  uint64  `db:"status"`
-	Package int64   `db:"package"`
+	Package uint64  `db:"package"`
 }
 
 type Comment struct {
-	Created time.Time `db:"created" json:"created"`
-	Updated time.Time `db:"updated" json:"updated,omitempty"`
-	Text    string    `db:"value" json:"text"`
-	ID      uint64    `db:"id" json:"id"`
+	Created time.Time `db:"created"`
+	Updated time.Time `db:"updated"`
+	Comment string    `db:"comment"`
+	ID      uint64    `db:"id"`
 }
 
 func NewComment(com *Comment) *pb.Comment {
 	return &pb.Comment{
 		Id:      com.ID,
-		Text:    com.Text,
+		Comment: com.Comment,
 		Created: timestamppb.New(com.Created),
 		Updated: timestamppb.New(com.Updated),
 	}

internal/service/client_git/client_test.go

@@ -1,81 +0,0 @@
-package client_git
-
-import (
-	"context"
-	"database/sql"
-	"embed"
-	"fmt"
-	"testing"
-
-	"git.unistack.org/unistack-org/pkgdash/internal/storage"
-	// "git.unistack.org/unistack-org/pkgdash/internal/storage/postgres"
-	"git.unistack.org/unistack-org/pkgdash/internal/storage/sqlite"
-	pb "git.unistack.org/unistack-org/pkgdash/proto"
-)
-
-func TestClientPG(t *testing.T) {
-	dsn := fmt.Sprintf("file:///database.db")
-	conn, err := sql.Open("sqlite", dsn)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer conn.Close()
-	if err = conn.Ping(); err != nil {
-		t.Fatal(err)
-	}
-
-	fucntion := sqlite.NewStorage()
-	st := fucntion(conn, embed.FS{})
-	s, ok := st.(storage.Storage)
-	if !ok {
-		t.Fatal("typecast error")
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	_ = cancel
-	cli := NewClient(1)
-
-	ch := cli.Run(ctx, s)
-
-	data := &pb.AddPackageReq{
-		Name: "test",
-		Url:  "https://github.com/dantedenis/service_history.git",
-	}
-
-	ch <- data
-
-	<-cli.Done()
-}
-
-func TestClientLite(t *testing.T) {
-	conn, err := sql.Open("sqlite3", "../../identifier.sqlite")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer conn.Close()
-	if err = conn.Ping(); err != nil {
-		t.Fatal(err)
-	}
-
-	function := sqlite.NewStorage()
-	st := function(conn, embed.FS{})
-	s, ok := st.(storage.Storage)
-	if !ok {
-		t.Fatal("typecast error")
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	_ = cancel
-	cli := NewClient(1)
-
-	ch := cli.Run(ctx, s)
-
-	data := &pb.AddPackageReq{
-		Name: "test",
-		Url:  "https://github.com/dantedenis/service_history.git",
-	}
-
-	ch <- data
-
-	<-cli.Done()
-}

internal/service/service.go

@@ -1,154 +0,0 @@
-package service
-
-import (
-	"context"
-	"database/sql"
-	"net/url"
-	"strings"
-
-	httpsrv "go.unistack.org/micro-server-http/v4" // TODO
-	"go.unistack.org/micro/v4"
-	"go.unistack.org/micro/v4/config"
-	microcfg "go.unistack.org/micro/v4/config"
-	"go.unistack.org/micro/v4/logger"
-	"go.unistack.org/micro/v4/options"
-	"go.unistack.org/micro/v4/register"
-	"go.unistack.org/micro/v4/server"
-	intcfg "git.unistack.org/unistack-org/pkgdash/config"
-	"git.unistack.org/unistack-org/pkgdash/handler"
-	pb "git.unistack.org/unistack-org/pkgdash/proto"
-	"git.unistack.org/unistack-org/pkgdash/service/client_git"
-	"git.unistack.org/unistack-org/pkgdash/storage"
-)
-
-func NewService(ctx context.Context) (micro.Service, error) {
-	var reg register.Register
-
-	cfg := intcfg.NewConfig(ServiceName, Service)
-
-	cs := microcfg.NewConfig(config.Struct(cfg))
-
-	// TODO
-	mgsrv := httpsrv.NewServer(
-		options.Register(reg),
-	)
-
-	svc := micro.NewService(
-		micro.Config(cs),
-	)
-
-	h := handler.NewHandler(svc, client_git.NewClient(5))
-
-	if err := svc.Init(
-		micro.AfterStart(func(_ context.Context) error {
-			return h.Init(svc.Options().Context)
-		}),
-		micro.BeforeStart(func(ctx context.Context) error {
-			if err := config.Load(ctx, []config.Config{cs}, config.LoadOverride(true)); err != nil {
-				return err
-			}
-			if err := config.Validate(ctx, cfg); err != nil {
-				return err
-			}
-
-			if err := svc.Init(
-				micro.Name(intcfg.ServiceName),
-				micro.Version(intcfg.ServiceVersion),
-			); err != nil {
-				return err
-			}
-
-			if err := svc.Server("http").Init(
-				options.Address(cfg.Address),
-				options.Name(cfg.App.Name),
-				server.Version(cfg.App.Version),
-			); err != nil {
-				return err
-			}
-
-			return nil
-		}),
-		micro.BeforeStart(func(_ context.Context) error {
-			log := logger.NewLogger(
-				logger.WithLevel(logger.ParseLevel(cfg.LogLevel)),
-				logger.WithCallerSkipCount(3),
-			)
-			return svc.Init(micro.Logger(log))
-		}),
-		micro.BeforeStart(func(ctx context.Context) error {
-			var connstr string
-			if v, ok := cfg.StorageDSN[cfg.App.Name]; ok {
-				connstr = v
-			} else if v, ok = cfg.StorageDSN["all"]; ok {
-				connstr = v
-			}
-
-			scheme, dsn, err := storageOptions(connstr)
-			if err != nil {
-				return err
-			}
-			conn, err := connectDataBase(scheme, dsn)
-			if err != nil {
-				return err
-			}
-			store, err := storage.NewStorage(scheme, conn)
-			if err != nil {
-				return err
-			}
-			ctx = storage.InContext(ctx, store)
-			return svc.Init(micro.Context(ctx))
-		}),
-	); err != nil {
-		return nil, err
-	}
-
-	if err := pb.RegisterPkgdashServiceServer(mgsrv, h); err != nil {
-		logger.Fatalf(ctx, "failed to register handler: %v", err)
-	}
-
-	intsvc := httpsrv.NewServer(
-		server.Codec("application/json", jsoncodec.NewCodec()),
-		server.Address(cfg.Meter.Addr), server.Context(ctx),
-	)
-
-	if err := intsvc.Init(); err != nil {
-		logger.Fatalf(ctx, "failed to init http srv: %v", err)
-	}
-
-	if err := healthhandler.RegisterHealthServiceServer(intsvc, healthhandler.NewHandler()); err != nil {
-		logger.Fatalf(ctx, "failed to set http handler: %v", err)
-	}
-	if err := meterhandler.RegisterMeterServiceServer(intsvc, meterhandler.NewHandler()); err != nil {
-		logger.Fatalf(ctx, "failed to set http handler: %v", err)
-	}
-
-	if err := intsvc.Start(); err != nil {
-		logger.Fatalf(ctx, "failed to run http srv: %v", err)
-	}
-
-	return svc, nil
-}
-
-func storageOptions(dsn string) (string, string, error) {
-	u, err := url.Parse(dsn)
-	if err != nil {
-		return "", "", err
-	}
-	scheme := u.Scheme
-	if idx := strings.Index(u.Scheme, "+"); idx > 0 {
-		scheme = u.Scheme[:idx]
-		u.Scheme = u.Scheme[idx+1:]
-	}
-	return scheme, u.String(), nil
-}
-
-func connectDataBase(driverName, dsn string) (*sql.DB, error) {
-	conn, err := sql.Open(driverName, dsn)
-	if err != nil {
-		return nil, err
-	}
-	if err = conn.Ping(); err != nil {
-		return nil, err
-	}
-	return conn, err
-}

internal/storage/migrations/sqlite/000001_init_schema.down.sql

@@ -1,5 +0,0 @@
-drop table if exists dashboard ;
-drop table if exists package ;
-drop table if exists module ;
-drop table if exists issue ;
-drop table if exists comment;

internal/storage/migrations/sqlite/000001_init_schema.up.sql

@@ -1,39 +0,0 @@
-create table if not exists dashboard (
-    id integer primary key autoincrement not null ,
-    "uuid" uuid not null unique ,
-    package integer[] default '{}'
-);
-
-create table if not exists comment (
-    id integer primary key autoincrement not null ,
-    "text" text ,
-    package integer not null,
-    created timestamp not null default current_timestamp ,
-    updated timestamp default current_timestamp
-);
-
-create table if not exists module (
-    id integer primary key autoincrement not null ,
-    name varchar not null ,
-    version varchar not null ,
-    last_version varchar not null
-);
-
-create table if not exists issue (
-    id serial not null unique primary key ,
-    --package integer references package(id) ,
-    modules integer[] default '[]',
-    status integer default 0 ,
-    "desc" varchar
-);
-
-create table if not exists package (
-    id integer primary key autoincrement not null ,
-    name varchar not null ,
-    url varchar ,
-    modules integer[] default '[]',
-    issues integer[] default '[]',
-    comments integer[] default '[]'
-);
-
-

internal/storage/sqlite/queries.go

@@ -1,34 +1,14 @@
 package sqlite
 
 const (
-	queryPackagesList = `select 
-    id,
-    name,
-    url,
-	comments
-	modules,
-	issues,
-    from package;
-`
-	queryCommentsCreate = `
-insert into comment(text) values ($1) returning id;
-update package
-set comments = json_insert(comments, '$[#]', ( select last_insert_rowid() as id from comment ))
-where id = $2 ;
-`
-	queryPackagesCreate = `
-insert into package(name, url, modules) values ($1, $2, $3);
-`
-	queryInsMsgGetIDs = `
-insert into module(name, version, last_version) values
-%s
-returning id;
-`
-	queryModulesList = `
-select id, name, version, last_version, created, updated from modules;
-`
-
-	queryCommentsList = `
-select id, text, created, updated from comments;
-`
+	queryPackagesProcess      = `select id, name, url, comments, modules, issues, created, updated from packages where ROUND((JULIANDAY(CURRENT_TIMESTAMP) - JULIANDAY(last_check)) * 86400) > $1 or last_check is NULL`
+	queryPackagesModulesCount = `update packages set modules = $2, last_check = CURRENT_TIMESTAMP where id = $1;`
+	queryPackagesList         = `select id, name, url, comments, modules, issues, created, updated from packages;`
+	queryPackagesLookup       = `select id, name, url, comments, modules, issues, created, updated from packages where id = $1;`
+	queryCommentsCreate       = `insert into comments (comment) values ($1) returning id;`
+	queryPackagesCreate       = `insert into packages (name, url) values ($1, $2) returning *;`
+	queryInsMsgGetIDs         = `insert into modules(name, version, last_version) values %s returning id;`
+	queryModulesList          = `select id, name, version, last_version, created, updated from modules;`
+	queryModulesCreate        = `insert into modules (name, version, last_version, package) values ($1, $2, $3, $4) returning *;`
+	queryCommentsList         = `select id, text, created, updated from comments;`
 )

internal/storage/sqlite/storage.go

@@ -2,130 +2,71 @@ package sqlite
 
 import (
 	"context"
-	"database/sql"
-	"embed"
-	"errors"
 	"fmt"
 	"strings"
+	"time"
 
-	"github.com/golang-migrate/migrate/v4"
-	"github.com/golang-migrate/migrate/v4/database/sqlite"
-	"github.com/golang-migrate/migrate/v4/source/iofs"
-	"github.com/lib/pq"
-	_ "github.com/mattn/go-sqlite3"
-	"go.unistack.org/micro/v4/logger"
 	"git.unistack.org/unistack-org/pkgdash/internal/models"
+	"git.unistack.org/unistack-org/pkgdash/internal/storage"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	"github.com/jmoiron/sqlx"
+	"go.unistack.org/micro/v4/logger"
 )
 
-const (
-	pathMigration = `migrations/sqlite`
-)
+func init() {
+	storage.RegisterStorage("sqlite", NewStorage())
+}
+
+var _ storage.Storage = (*Sqlite)(nil)
 
 type Sqlite struct {
-	db *sql.DB
-	fs embed.FS
+	db *sqlx.DB
 }
 
-func NewStorage() func(*sql.DB, embed.FS) interface{} {
-	return func(db *sql.DB, fs embed.FS) interface{} {
-		return &Sqlite{db: db, fs: fs}
+func NewStorage() func(*sqlx.DB) interface{} {
+	return func(db *sqlx.DB) interface{} {
+		return &Sqlite{db: db}
 	}
 }
 
-func (s *Sqlite) MigrateUp() error {
-	driver, err := sqlite.WithInstance(s.db, &sqlite.Config{
-		MigrationsTable: sqlite.DefaultMigrationsTable,
-		DatabaseName:    "pkgdash",
-	})
-	if err != nil {
-		return err
-	}
-	source, err := iofs.New(s.fs, pathMigration)
-	if err != nil {
-		return err
-	}
-
-	// TODO: pass own logger
-	m, err := migrate.NewWithInstance("fs", source, "pkgdash", driver)
-	if err != nil {
-		return err
-	}
-
-	if err = m.Up(); err != nil && !errors.Is(err, migrate.ErrNoChange) {
-		return err
-	}
-
-	return nil
+func (s *Sqlite) PackagesDelete(ctx context.Context, req *pb.PackagesDeleteReq) error {
+	return fmt.Errorf("need implement")
 }
 
-func (s *Sqlite) MigrateDown() error {
-	driver, err := sqlite.WithInstance(s.db, &sqlite.Config{
-		MigrationsTable: sqlite.DefaultMigrationsTable,
-		DatabaseName:    "pkgdash",
-	})
-	if err != nil {
-		return err
-	}
-	source, err := iofs.New(s.fs, pathMigration)
-	if err != nil {
-		return err
-	}
-
-	// TODO: pass own logger
-	m, err := migrate.NewWithInstance("fs", source, "pkgdash", driver)
-	if err != nil {
-		return err
-	}
-
-	if err = m.Down(); err != nil && !errors.Is(err, migrate.ErrNoChange) {
-		return err
-	}
-
-	return nil
+func (s *Sqlite) PackagesUpdate(ctx context.Context, req *pb.PackagesUpdateReq) (*models.Package, error) {
+	return nil, fmt.Errorf("need implement")
 }
 
-func (s *Sqlite) PackagesUpdate(ctx context.Context, req *pb.PackagesUpdateReq) error {
-	panic("need implement")
+func (s *Sqlite) PackagesLookup(ctx context.Context, req *pb.PackagesLookupReq) (*models.Package, error) {
+	pkg := &models.Package{}
+
+	err := s.db.GetContext(ctx, pkg, queryPackagesLookup, req.Id)
+	if err != nil {
+		return nil, err
+	}
+
+	return pkg, err
 }
 
 func (s *Sqlite) PackagesList(ctx context.Context, req *pb.PackagesListReq) ([]*models.Package, error) {
 	var packages []*models.Package
 
-	rows, err := s.db.QueryContext(ctx, queryPackagesList)
+	err := s.db.SelectContext(ctx, &packages, queryPackagesList)
 	if err != nil {
 		return nil, err
 	}
 
-	for ; rows.Err() == nil; rows.Next() {
-		pkg := &models.Package{}
-		if err = rows.Scan(
-			&pkg.ID,
-			&pkg.Name,
-			&pkg.URL,
-			&pkg.Comments,
-		); err != nil {
-			_ = rows.Close()
-			return nil, err
-		}
-		packages = append(packages, pkg)
-	}
-
-	if err = rows.Err(); err != nil {
-		return nil, err
-	}
-
-	if err = rows.Close(); err != nil {
-		return nil, err
-	}
-
-	return packages, err
+	return packages, nil
 }
 
-func (s *Sqlite) CommentsCreate(ctx context.Context, req *pb.CommentsCreateReq) (id uint64, err error) {
+func (s *Sqlite) CommentsDelete(ctx context.Context, req *pb.CommentsDeleteReq) error {
+	return nil
+}
+
+func (s *Sqlite) CommentsCreate(ctx context.Context, req *pb.CommentsCreateReq) (*models.Comment, error) {
 	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
-		return 0, err
+		return nil, err
 	}
 
 	defer func() {
@@ -138,85 +79,61 @@ func (s *Sqlite) CommentsCreate(ctx context.Context, req *pb.CommentsCreateReq)
 		}
 	}()
 
-	if err = tx.QueryRowContext(ctx, queryCommentsCreate, req.Text, req.PackageId).Scan(&id); err != nil {
-		return id, err
+	if _, err = tx.ExecContext(ctx, queryCommentsCreate, req.Comment, req.PackageId); err != nil {
+		return nil, err
 	}
 
-	return id, err
+	return nil, nil
 }
 
-func (s *Sqlite) PackagesCreate(ctx context.Context, req *pb.PackagesCreateReq) error {
-	tx, err := s.db.BeginTx(ctx, nil)
+func (s *Sqlite) PackagesProcess(ctx context.Context, td time.Duration) ([]*models.Package, error) {
+	var packages []*models.Package
+	err := s.db.SelectContext(ctx, &packages, queryPackagesProcess, td.Seconds())
+	if err != nil {
+		return nil, err
+	}
+
+	return packages, nil
+}
+
+func (s *Sqlite) PackagesCreate(ctx context.Context, req *pb.PackagesCreateReq) (*models.Package, error) {
+	pkg := &models.Package{}
+	err := s.db.GetContext(ctx, pkg, queryPackagesCreate, req.Name, req.Url)
+	if err != nil {
+		return nil, err
+	}
+
+	return pkg, nil
+}
+
+func (s *Sqlite) PackagesModulesCreate(ctx context.Context, pkg *models.Package, modules []*models.Module) error {
+	tx, err := s.db.BeginTxx(ctx, nil)
 	if err != nil {
 		return err
 	}
 
-	defer func() {
+	for _, mod := range modules {
+		err = tx.GetContext(ctx, mod, queryModulesCreate, mod.Name, mod.Version, mod.LastVersion, mod.Package)
 		if err != nil {
-			if rollbackErr := tx.Rollback(); rollbackErr != nil {
-				logger.Errorf(ctx, "AddPackage: unable to rollback: %v", rollbackErr)
-			}
-		} else {
-			err = tx.Commit()
+			_ = tx.Rollback()
+			return err
 		}
-	}()
+	}
 
-	res, err := tx.ExecContext(ctx, queryPackagesCreate, req.Name, req.Url, pq.Array(req.Modules))
-	if err != nil {
+	if _, err = tx.ExecContext(ctx, queryPackagesModulesCount, pkg.ID, len(modules)); err != nil {
+		_ = tx.Rollback()
 		return err
 	}
 
-	if aff, affErr := res.RowsAffected(); err != nil {
-		err = affErr
-	} else if aff == 0 {
-		err = errors.New("rows affected is 0")
+	if err = tx.Commit(); err != nil {
+		_ = tx.Rollback()
+		return err
 	}
 
-	return err
+	return nil
 }
 
-func (s *Sqlite) InsertButchModules(ctx context.Context, req []models.Module) ([]uint64, error) {
-	tx, err := s.db.BeginTx(ctx, nil)
-	if err != nil {
-		return nil, err
-	}
-	defer func() {
-		if err != nil {
-			if rollbackErr := tx.Rollback(); rollbackErr != nil {
-				logger.Errorf(ctx, "AddPackage: unable to rollback: %v", rollbackErr)
-			}
-		} else {
-			err = tx.Commit()
-		}
-	}()
-
-	query := generateQuery(req)
-
-	rows, err := tx.QueryContext(ctx, query)
-	if err != nil {
-		return nil, err
-	}
-	defer func() {
-		if err = rows.Close(); err != nil {
-			return
-		}
-		err = rows.Err()
-	}()
-
-	result := make([]uint64, 0)
-	for rows.Next() {
-		tmp := uint64(0)
-		if err = rows.Scan(&tmp); err != nil {
-			return nil, err
-		}
-
-		result = append(result, tmp)
-	}
-
-	return result, err
-}
-
-func (s *Sqlite) GetModule(ctx context.Context, req *pb.ModulesListReq) ([]*models.Module, error) {
+func (s *Sqlite) ModulesList(ctx context.Context, req *pb.ModulesListReq) ([]*models.Module, error) {
 	var err error
 	var modules []*models.Module
 
@@ -259,50 +176,14 @@ func (s *Sqlite) GetModule(ctx context.Context, req *pb.ModulesListReq) ([]*mode
 func (s *Sqlite) CommentsList(ctx context.Context, req *pb.CommentsListReq) ([]*models.Comment, error) {
 	var comments []*models.Comment
 
-	rows, err := s.db.QueryContext(ctx, queryCommentsList, req.PackageId)
+	err := s.db.SelectContext(ctx, &comments, queryCommentsList, req.PackageId)
 	if err != nil {
 		return nil, err
 	}
-	defer func() {
-		if err = rows.Close(); err != nil {
-			return
-		}
-		err = rows.Err()
-	}()
-	for ; rows.Err() == nil; rows.Next() {
-		com := &models.Comment{}
-		if err = rows.Scan(
-			&com.ID,
-			&com.Text,
-			&com.Created,
-			&com.Updated,
-		); err != nil {
-			_ = rows.Close()
-			return nil, err
-		}
-
-		comments = append(comments, com)
-	}
-
-	if err = rows.Err(); err != nil {
-		return nil, err
-	}
-
-	if err = rows.Close(); err != nil {
-		return nil, err
-	}
 
 	return comments, nil
 }
 
-func convertSliceUInt(arg ...uint64) []interface{} {
-	result := make([]interface{}, 0, len(arg))
-	for i := range arg {
-		result = append(result, arg[i])
-	}
-	return result
-}
-
 func generateQuery(rsp []models.Module) string {
 	const pattern = `%c('%s', '%s', '%s')`
 	build := strings.Builder{}
@@ -315,7 +196,3 @@ func generateQuery(rsp []models.Module) string {
 
 	return fmt.Sprintf(queryInsMsgGetIDs, build.String())
 }
-
-func generateArrayIneq(count int) string {
-	return "(?" + strings.Repeat(",?", count-1) + ")"
-}

internal/storage/storage.go

@@ -2,68 +2,43 @@ package storage
 
 import (
 	"context"
-	"database/sql"
-	"embed"
 	"errors"
+	"time"
 
 	"git.unistack.org/unistack-org/pkgdash/internal/models"
-	// "git.unistack.org/unistack-org/pkgdash/internal/storage/postgres"
-
-	"git.unistack.org/unistack-org/pkgdash/internal/storage/sqlite"
 	pb "git.unistack.org/unistack-org/pkgdash/proto"
+	"github.com/jmoiron/sqlx"
 )
 
-//go:embed migrations
-var fs embed.FS
-
-var storages = map[string]func(*sql.DB, embed.FS) interface{}{
-	//"postgres": postgres.NewStorage(),
-	"sqlite": sqlite.NewStorage(),
+func RegisterStorage(name string, fn func(*sqlx.DB) interface{}) {
+	storages[name] = fn
 }
 
-type contextKey string
-
-var storeIdent = contextKey("store")
-
-type Migrate interface {
-	MigrateUp() error
-	MigrateDown() error
-}
+var storages = map[string]func(*sqlx.DB) interface{}{}
 
 type Storage interface {
-	Migrate
-	PackagesCreate(ctx context.Context, req *pb.PackagesCreateReq) error
+	PackagesProcess(ctx context.Context, td time.Duration) ([]*models.Package, error)
+	PackagesCreate(ctx context.Context, req *pb.PackagesCreateReq) (*models.Package, error)
 	PackagesList(ctx context.Context, req *pb.PackagesListReq) ([]*models.Package, error)
-	PackagesUpdate(ctx context.Context, req *pb.PackagesUpdateReq) error
+	PackagesLookup(ctx context.Context, req *pb.PackagesLookupReq) (*models.Package, error)
+	PackagesUpdate(ctx context.Context, req *pb.PackagesUpdateReq) (*models.Package, error)
 	PackagesDelete(ctx context.Context, req *pb.PackagesDeleteReq) error
 	CommentsCreate(ctx context.Context, req *pb.CommentsCreateReq) (*models.Comment, error)
 	CommentsDelete(ctx context.Context, req *pb.CommentsDeleteReq) error
 	CommentsList(ctx context.Context, req *pb.CommentsListReq) ([]*models.Comment, error)
-	InsertButchModules(ctx context.Context, req []models.Module) ([]uint64, error)
 	ModulesList(ctx context.Context, req *pb.ModulesListReq) ([]*models.Module, error)
+	PackagesModulesCreate(ctx context.Context, pkg *models.Package, modules []*models.Module) error
 }
 
-func NewStorage(name string, db *sql.DB) (Storage, error) {
+func NewStorage(name string, db *sqlx.DB) (Storage, error) {
 	function, ok := storages[name]
 	if !ok {
 		return nil, errors.New("incorrect name store")
 	}
-	store := function(db, fs)
+	store := function(db)
 	database, ok := store.(Storage)
 	if !ok {
 		return nil, errors.New("dont implements interface Storage")
 	}
 	return database, nil
 }
-
-func InContext(ctx context.Context, val Storage) context.Context {
-	return context.WithValue(ctx, storeIdent, val)
-}
-
-func FromContext(ctx context.Context) (Storage, error) {
-	if store, ok := ctx.Value(storeIdent).(Storage); !ok {
-		return nil, errors.New("empty store")
-	} else {
-		return store, nil
-	}
-}

internal/worker/worker.go

@@ -1,96 +1,76 @@
-package client_git
+package worker
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"io"
 	"net/url"
 	"os"
 	"sort"
 	"strings"
+	"sync"
+	"time"
 
+	"git.unistack.org/unistack-org/pkgdash/internal"
+	"git.unistack.org/unistack-org/pkgdash/internal/models"
+	"git.unistack.org/unistack-org/pkgdash/internal/storage"
 	"github.com/go-git/go-git/v5"
 	"github.com/go-git/go-git/v5/plumbing/filemode"
 	"github.com/go-git/go-git/v5/plumbing/object"
 	"github.com/go-git/go-git/v5/storage/memory"
 	"github.com/pkg/errors"
 	"go.unistack.org/micro/v4/logger"
-	"git.unistack.org/unistack-org/pkgdash/internal"
-	"git.unistack.org/unistack-org/pkgdash/internal/models"
-	"git.unistack.org/unistack-org/pkgdash/internal/storage"
-	pb "git.unistack.org/unistack-org/pkgdash/proto"
 	"golang.org/x/mod/modfile"
 	"golang.org/x/mod/module"
 )
 
-type Client interface {
-	Run(ctx context.Context, st storage.Storage) chan *pb.PackagesCreateReq
-	IsClose() bool
-	Done() <-chan struct{}
-}
-
-type client struct {
-	worker chan *pb.PackagesCreateReq
-	closed bool
-	lock   chan struct{}
-}
-
-func NewClient(cap uint) Client {
-	return &client{
-		make(chan *pb.PackagesCreateReq, cap),
-		false,
-		make(chan struct{}),
-	}
-}
-
-func (c *client) Run(ctx context.Context, st storage.Storage) chan *pb.PackagesCreateReq {
-	go func() {
-		defer close(c.worker)
-		for {
-			select {
-			case data := <-c.worker:
-				go func() {
-					runner(ctx, st, data)
-				}()
-			case <-ctx.Done():
-				logger.Info(ctx, ctx.Err())
-				return
+func Run(ctx context.Context, store storage.Storage, td time.Duration) {
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+	var wg sync.WaitGroup
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			logger.Infof(ctx, "check packages to process")
+			packages, err := store.PackagesProcess(ctx, td)
+			logger.Infof(ctx, "check packages to process %#+v, err: %v", packages, err)
+			if err != nil && err != sql.ErrNoRows {
+				logger.Fatalf(ctx, "failed to get packages to process: %v", err)
 			}
+			wg.Add(len(packages))
+			for _, pkg := range packages {
+				go func(p *models.Package) {
+					if err := process(ctx, store, p); err != nil {
+						logger.Errorf(ctx, "failed to process package %s: %v", p.Name, err)
+					}
+					wg.Done()
+				}(pkg)
+			}
+			wg.Wait()
 		}
-	}()
-
-	return c.worker
+	}
 }
 
-func (c *client) IsClose() bool {
-	return c.closed
-}
-
-// Done for locked goroutine
-func (c *client) Done() <-chan struct{} {
-	return c.lock
-}
-
-func runner(ctx context.Context, st storage.Storage, req *pb.PackagesCreateReq) {
-	modules, err := getGoModule(ctx, req.Url)
+func process(ctx context.Context, store storage.Storage, pkg *models.Package) error {
+	logger.Infof(ctx, "process package %v", pkg)
+	modules, err := getGoModule(ctx, pkg.ID, pkg.URL)
 	if err != nil {
-		logger.Error(ctx, err)
-		return
+		logger.Errorf(ctx, "failed to get modules: %v", err)
+		return err
 	}
 
-	logger.Infof(ctx, "success get list mod", modules)
-
-	if req.Modules, err = st.InsertButchModules(ctx, modules); err != nil {
-		logger.Error(ctx, err)
-		return
+	if err = store.PackagesModulesCreate(ctx, pkg, modules); err != nil {
+		logger.Errorf(ctx, "failed to set create modules: %v", err)
+		return err
 	}
 
-	if err = st.PackagesCreate(ctx, req); err != nil {
-		logger.Error(ctx, err)
-	}
+	return nil
 }
 
-func getGoModule(ctx context.Context, gitUrl string) ([]models.Module, error) {
+func getGoModule(ctx context.Context, pkgID uint64, gitUrl string) ([]*models.Module, error) {
 	u, err := url.Parse(gitUrl)
 	if err != nil {
 		logger.Fatal(ctx, err)
@@ -135,7 +115,7 @@ func getGoModule(ctx context.Context, gitUrl string) ([]models.Module, error) {
 		return nil, err
 	}
 
-	unique := make(map[string]models.Module)
+	unique := make(map[string]*models.Module)
 	var mvs []module.Version
 	err = tree.Files().ForEach(func(file *object.File) error {
 		if file == nil {
@@ -151,7 +131,8 @@ func getGoModule(ctx context.Context, gitUrl string) ([]models.Module, error) {
 					return err
 				}
 				for i := range mvs {
-					unique[mvs[i].Path] = models.Module{
+					unique[mvs[i].Path] = &models.Module{
+						Package:     pkgID,
 						Name:        mvs[i].Path,
 						Version:     mvs[i].Version,
 						LastVersion: mvs[i].Version,
@@ -177,7 +158,7 @@ func getGoModule(ctx context.Context, gitUrl string) ([]models.Module, error) {
 		return nil
 	})
 
-	result := make([]models.Module, 0, len(unique))
+	result := make([]*models.Module, 0, len(unique))
 	for _, v := range unique {
 		result = append(result, v)
 	}