adding inserts with duplicate ref_id, func skipping (#3)

* adding inserts with duplicate ref_id, func skipping

* remove comment

Author: agatons

dao.go

@@ -54,6 +54,45 @@ func (d *docket) initTables() error {
 // InsertTasks - supports max 65535/5 = 13,107 tasks in one batch, since postgres supports max 65535 arguments per statement
 // if you need more, execute multiple InsertTasks in the same transaction
 func (d *docket) InsertTasks(tx *sql.Tx, tcs ...TaskCreator) ([]Task, error) {
+	return d.insertTasks(tx, false, tcs...)
+}
+
+func (d *docket) InsertTask(tx *sql.Tx, tc TaskCreator) (Task, error) {
+	tasks, err := d.InsertTasks(tx, tc)
+	if err != nil {
+		return nil, err
+	}
+	if len(tasks) == 1 {
+		return tasks[0], nil
+	}
+	return nil, errors.New("shouldn't happen, got wrong number of tasks from InsertTasks, this is a bug")
+}
+
+// InsertTasksSkipDuplicates - supports max 65535/5 = 13,107 tasks in one batch, since postgres supports max 65535 arguments per statement
+// if you need more, execute multiple InsertTasks in the same transaction
+// requires ref_id to be set on all tasks
+func (d *docket) InsertTasksSkipDuplicates(tx *sql.Tx, tcs ...TaskCreator) ([]Task, error) {
+	for _, tc := range tcs {
+		if !tc.task.refId.Valid {
+			return nil, errors.New("InsertTasksSkipDuplicates requires ref_id to be set on all tasks")
+		}
+	}
+	return d.insertTasks(tx, true, tcs...)
+}
+
+func (d *docket) InsertTaskSkipDuplicates(tx *sql.Tx, tc TaskCreator) (Task, error) {
+	tasks, err := d.InsertTasksSkipDuplicates(tx, tc)
+	if err != nil {
+		return nil, err
+	}
+	if len(tasks) == 1 {
+		return tasks[0], nil
+	}
+	// allowed to skip insert if duplicate, so return nil
+	return nil, nil
+}
+
+func (d *docket) insertTasks(tx *sql.Tx, skipDuplicates bool, tcs ...TaskCreator) ([]Task, error) {
 	shouldNotify := false
 	var placeholders []string
 	var args []interface{}
@@ -75,11 +114,16 @@ func (d *docket) InsertTasks(tx *sql.Tx, tcs ...TaskCreator) ([]Task, error) {
 	q := `
 		INSERT INTO pqdocket_task(scheduled_at, ref_id, func, metadata, claim_time_seconds)
 		VALUES %s
+		--DUPLICATES_HANDLING--
 		RETURNING *
 	`
 	q = fmt.Sprintf(q, strings.Join(placeholders, ", "))
 	q = strings.Replace(q, taskTableName, d.tableName(), 1)
 
+	if skipDuplicates {
+		q = strings.Replace(q, "--DUPLICATES_HANDLING--", "ON CONFLICT (ref_id, func) DO NOTHING", 1)
+	}
+
 	var rows *sql.Rows
 	var err error
 	if tx == nil {
@@ -110,17 +154,6 @@ func (d *docket) InsertTasks(tx *sql.Tx, tcs ...TaskCreator) ([]Task, error) {
 	return tasks, rows.Close()
 }
 
-func (d *docket) InsertTask(tx *sql.Tx, tc TaskCreator) (Task, error) {
-	tasks, err := d.InsertTasks(tx, tc)
-	if err != nil {
-		return nil, err
-	}
-	if len(tasks) == 1 {
-		return tasks[0], nil
-	}
-	return nil, errors.New("shouldn't happen, got wrong number of tasks from InsertTasks, this is a bug")
-}
-
 func (d *docket) claimTasks(wantNum int) ([]task, error) {
 	q := `
 	WITH tasks_to_claim AS (

docket.go

@@ -43,6 +43,12 @@ type Docket interface {
 	// If you need more, execute multiple InsertTasks in the same transaction.
 	InsertTasks(tx *sql.Tx, tc ...TaskCreator) ([]Task, error)
 
+	// InsertTaskSkipDuplicates - insert a task. Tasks are immediately available to run after being inserted. If a task with the same (func, ref_id) exists, the insert is skipped and no error is returned. Cannot be used without ref id.
+	InsertTaskSkipDuplicates(tx *sql.Tx, tc TaskCreator) (Task, error)
+
+	// InsertTasksSkipDuplicates - batch insert tasks, skipping tasks with duplicate (func, ref_id) Cannot be used without ref id.
+	InsertTasksSkipDuplicates(tx *sql.Tx, tc ...TaskCreator) ([]Task, error)
+
 	// FindTasks is used to query tasks.
 	// If executed within a transaction (tx != nil), it will lock the returned rows for the duration of the transaction.
 	FindTasks(tx *sql.Tx, params ...FindParam) ([]Task, error)