feat: refactor queue module

.github/workflows/pr-check-title.yml

@@ -1,6 +1,10 @@
 name: PR Check Title
 on:
-  pull_request:
+  pull_request_target:
+    types:
+      - opened
+      - edited
+      - synchronize
 jobs:
   pr-check-title:
     runs-on: ubuntu-latest

contracts/queue/driver.go

@@ -0,0 +1,16 @@
+package queue
+
+type Driver interface {
+	// Connection returns the connection name for the driver.
+	Connection() string
+	// Driver returns the driver name for the driver.
+	Driver() string
+	// Push pushes the job onto the queue.
+	Push(job Job, args []any, queue string) error
+	// Bulk pushes a slice of jobs onto the queue.
+	Bulk(jobs []Jobs, queue string) error
+	// Later pushes the job onto the queue after a delay.
+	Later(delay uint, job Job, args []any, queue string) error
+	// Pop pops the next job off of the queue.
+	Pop(queue string) (Job, []any, error)
+}

contracts/queue/job.go

@@ -8,6 +8,7 @@ type Job interface {
 }
 
 type Jobs struct {
-	Job  Job
-	Args []Arg
+	Job   Job
+	Args  []any
+	Delay uint
 }

contracts/queue/queue.go

@@ -1,19 +1,22 @@
 package queue
 
 type Queue interface {
-	Worker(args *Args) Worker
+	Worker(payloads ...*Args) Worker
 	// Register register jobs
-	Register(jobs []Job)
+	Register(jobs []Job) error
 	// GetJobs get all jobs
 	GetJobs() []Job
+	// GetJob get job by signature
+	GetJob(signature string) (Job, error)
 	// Job add a job to queue
-	Job(job Job, args []Arg) Task
+	Job(job Job, args []any) Task
 	// Chain creates a chain of jobs to be processed one by one, passing
 	Chain(jobs []Jobs) Task
 }
 
 type Worker interface {
 	Run() error
+	Shutdown() error
 }
 
 type Args struct {
@@ -24,8 +27,3 @@ type Args struct {
 	// Concurrent num
 	Concurrent int
 }
-
-type Arg struct {
-	Type  string
-	Value any
-}

contracts/queue/task.go

@@ -1,16 +1,12 @@
 package queue
 
-import (
-	"time"
-)
-
 type Task interface {
 	// Dispatch dispatches the task.
 	Dispatch() error
 	// DispatchSync dispatches the task synchronously.
 	DispatchSync() error
-	// Delay dispatches the task after the given delay.
-	Delay(time time.Time) Task
+	// Delay dispatches the task after the given time.
+	Delay(time uint) Task
 	// OnConnection sets the connection of the task.
 	OnConnection(connection string) Task
 	// OnQueue sets the queue of the task.

database/console/migrate_stubs.go

@@ -1,14 +1,13 @@
 package console
 
-type MysqlStubs struct {
-}
+type MysqlStubs struct{}
 
 // CreateUp Create up migration content.
 func (receiver MysqlStubs) CreateUp() string {
 	return `CREATE TABLE DummyTable (
-  id bigint(20) unsigned NOT NULL AUTO_INCREMENT,
-  created_at datetime(3) NOT NULL,
-  updated_at datetime(3) NOT NULL,
+  id BIGINT(20) UNSIGNED NOT NULL AUTO_INCREMENT,
+  created_at DATETIME(3) NOT NULL,
+  updated_at DATETIME(3) NOT NULL,
   PRIMARY KEY (id),
   KEY idx_DummyTable_created_at (created_at),
   KEY idx_DummyTable_updated_at (updated_at)
@@ -24,7 +23,7 @@
 
 // UpdateUp Update up migration content.
 func (receiver MysqlStubs) UpdateUp() string {
-	return `ALTER TABLE DummyTable ADD column varchar(255) COMMENT '';
+	return `ALTER TABLE DummyTable ADD column VARCHAR(255);
 `
 }
 
@@ -34,16 +33,18 @@
 `
 }
 
-type PostgresqlStubs struct {
-}
+type PostgresqlStubs struct{}
 
 // CreateUp Create up migration content.
 func (receiver PostgresqlStubs) CreateUp() string {
 	return `CREATE TABLE DummyTable (
-  id SERIAL PRIMARY KEY NOT NULL,
-  created_at timestamp NOT NULL,
-  updated_at timestamp NOT NULL
+  id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY NOT NULL,
+  created_at TIMESTAMP NOT NULL,
+  updated_at TIMESTAMP NOT NULL
 );
+
+CREATE INDEX idx_DummyTable_created_at ON DummyTable (created_at);
+CREATE INDEX idx_DummyTable_updated_at ON DummyTable (updated_at);
 `
 }
 
@@ -55,7 +56,7 @@
 
 // UpdateUp Update up migration content.
 func (receiver PostgresqlStubs) UpdateUp() string {
-	return `ALTER TABLE DummyTable ADD column varchar(255) NOT NULL;
+	return `ALTER TABLE DummyTable ADD column TEXT NOT NULL;
 `
 }
 
@@ -65,16 +66,18 @@
 `
 }
 
-type SqliteStubs struct {
-}
+type SqliteStubs struct{}
 
 // CreateUp Create up migration content.
 func (receiver SqliteStubs) CreateUp() string {
 	return `CREATE TABLE DummyTable (
-  id integer PRIMARY KEY AUTOINCREMENT NOT NULL,
-  created_at datetime NOT NULL,
-  updated_at datetime NOT NULL
+  id BIGINT PRIMARY KEY AUTOINCREMENT NOT NULL,
+  created_at DATETIME NOT NULL,
+  updated_at DATETIME NOT NULL
 );
+
+CREATE INDEX idx_DummyTable_created_at ON DummyTable (created_at);
+CREATE INDEX idx_DummyTable_updated_at ON DummyTable (updated_at);
 `
 }
 
@@ -86,7 +89,7 @@
 
 // UpdateUp Update up migration content.
 func (receiver SqliteStubs) UpdateUp() string {
-	return `ALTER TABLE DummyTable ADD column text;
+	return `ALTER TABLE DummyTable ADD column TEXT;
 `
 }
 
@@ -96,17 +99,19 @@
 `
 }
 
-type SqlserverStubs struct {
-}
+type SqlserverStubs struct{}
 
 // CreateUp Create up migration content.
 func (receiver SqlserverStubs) CreateUp() string {
 	return `CREATE TABLE DummyTable (
-  id bigint NOT NULL IDENTITY(1,1),
-  created_at datetime NOT NULL,
-  updated_at datetime NOT NULL,
+  id BIGINT NOT NULL IDENTITY(1,1),
+  created_at DATETIME NOT NULL,
+  updated_at DATETIME NOT NULL,
   PRIMARY KEY (id)
 );
+
+CREATE INDEX idx_DummyTable_created_at ON DummyTable (created_at);
+CREATE INDEX idx_DummyTable_updated_at ON DummyTable (updated_at);
 `
 }
 
@@ -118,7 +123,7 @@
 
 // UpdateUp Update up migration content.
 func (receiver SqlserverStubs) UpdateUp() string {
-	return `ALTER TABLE DummyTable ADD column varchar(255);
+	return `ALTER TABLE DummyTable ADD column VARCHAR(255);
 `
 }
 

event/application.go

@@ -30,7 +30,10 @@
 		}
 	}
 
-	app.queue.Register(jobs)
+	err := app.queue.Register(jobs)
+	if err != nil {
+		panic(err.Error())
+	}
 }
 
 func (app *Application) GetEvents() map[event.Event][]event.Listener {

event/service_provider.go

@@ -3,7 +3,7 @@
 import (
 	"github.com/goravel/framework/contracts/console"
 	"github.com/goravel/framework/contracts/foundation"
-	eventConsole "github.com/goravel/framework/event/console"
+	eventconsole "github.com/goravel/framework/event/console"
 )
 
 const Binding = "goravel.event"
@@ -23,7 +23,7 @@
 
 func (receiver *ServiceProvider) registerCommands(app foundation.Application) {
 	app.MakeArtisan().Register([]console.Command{
-		&eventConsole.EventMakeCommand{},
-		&eventConsole.ListenerMakeCommand{},
+		&eventconsole.EventMakeCommand{},
+		&eventconsole.ListenerMakeCommand{},
 	})
 }

event/task.go

@@ -62,13 +62,10 @@ func (receiver *Task) Dispatch() error {
 	return nil
 }
 
-func eventArgsToQueueArgs(args []event.Arg) []queuecontract.Arg {
-	var queueArgs []queuecontract.Arg
+func eventArgsToQueueArgs(args []event.Arg) []any {
+	var queueArgs []any
 	for _, arg := range args {
-		queueArgs = append(queueArgs, queuecontract.Arg{
-			Type:  arg.Type,
-			Value: arg.Value,
-		})
+		queueArgs = append(queueArgs, arg.Value)
 	}
 
 	return queueArgs

event/task_test.go

@@ -7,7 +7,6 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/goravel/framework/contracts/event"
-	queuecontract "github.com/goravel/framework/contracts/queue"
 	queuemock "github.com/goravel/framework/mocks/queue"
 )
 
@@ -32,9 +31,7 @@ func TestDispatch(t *testing.T) {
 				listener := &TestListener{}
 				mockTask := &queuemock.Task{}
 
-				mockQueue.On("Job", listener, []queuecontract.Arg{
-					{Type: "string", Value: "test"},
-				}).Return(mockTask).Once()
+				mockQueue.On("Job", listener, []any{"test"}).Return(mockTask).Once()
 				mockTask.On("DispatchSync").Return(nil).Once()
 
 				task = NewTask(mockQueue, []event.Arg{
@@ -51,9 +48,7 @@ func TestDispatch(t *testing.T) {
 				listener := &TestListenerHandleError{}
 				mockTask := &queuemock.Task{}
 
-				mockQueue.On("Job", listener, []queuecontract.Arg{
-					{Type: "string", Value: "test"},
-				}).Return(mockTask).Once()
+				mockQueue.On("Job", listener, []any{"test"}).Return(mockTask).Once()
 				mockTask.On("DispatchSync").Return(errors.New("error")).Once()
 
 				task = NewTask(mockQueue, []event.Arg{