Files
ragflow/internal/entity/base.go
Jin Hai e96bc37d06 Go: use NATS as the message queue (#15327)
### What problem does this PR solve?

```
RAGFlow(admin)> mq publish 'msg2';
SUCCESS
RAGFlow(admin)> mq publish 'msg3';
SUCCESS
RAGFlow(admin)> mq list;
+---------+---------------+
| message | subject       |
+---------+---------------+
| msg1    | tasks.RAGFLOW |
| msg2    | tasks.RAGFLOW |
| msg3    | tasks.RAGFLOW |
+---------+---------------+
RAGFlow(admin)> mq pull 2;
+---------+---------------+
| message | subject       |
+---------+---------------+
| msg1    | tasks.RAGFLOW |
| msg2    | tasks.RAGFLOW |
+---------+---------------+
RAGFlow(admin)> mq pull noack;
+---------+---------------+
| message | subject       |
+---------+---------------+
| abc     | tasks.RAGFLOW |
+---------+---------------+
RAGFlow(admin)> mq show
+-------------------+----------------+--------+---------------+---------------+-------------------+---------------+
| ack_pending_count | consumer_count | memory | message_count | pending_count | redelivered_count | waiting_count |
+-------------------+----------------+--------+---------------+---------------+-------------------+---------------+
| 2                 | 1              | 0      | 2             | 0             | 1                 | 0             |
+-------------------+----------------+--------+---------------+---------------+-------------------+---------------+

RAGFlow(admin)> list ingestors;
+--------------+-------------------------------------------+--------+
| host         | name                                      | status |
+--------------+-------------------------------------------+--------+
| 192.168.1.38 | ingestor-8f0e4bd5650a4ac58b0151969fbf6935 | alive  |
+--------------+-------------------------------------------+--------+

RAGFlow(admin)> list ingestion tasks;
+----------------------------------+----------------------------------+-----------+------+-------------+----------------------------------+
| document_id                      | id                               | status    | step | user        | user_id                          |
+----------------------------------+----------------------------------+-----------+------+-------------+----------------------------------+
| ffe64fae423411f1a2d938a74640adcc | 90d3d0f6528941c1ac8eb0360effccc4 | COMPLETED | 5    | aaa@aaa.com | 2ba4881420fa11f19e9c38a74640adcc |
+----------------------------------+----------------------------------+-----------+------+-------------+----------------------------------+

RAGFlow(admin)> remove ingestion tasks '90d3d0f6528941c1ac8eb0360effccc4';
+---------+----------------------------------+
| delete  | task_id                          |
+---------+----------------------------------+
| success | 90d3d0f6528941c1ac8eb0360effccc4 |
+---------+----------------------------------+

RAGFlow(admin)> stop ingestion tasks 'e89e20d9a25848a1b79bd9345ddbfe1d';
+----------+----------------------------------+
| status   | task_id                          |
+----------+----------------------------------+
| STOPPING | e89e20d9a25848a1b79bd9345ddbfe1d |
+----------+----------------------------------+

# Publish a message
RAGFlow(admin)> mq publish 'cdd';
SUCCESS

# List current tasks in the message queue
RAGFlow(admin)> mq list
+----------------------------------+---------------+
| message                          | subject       |
+----------------------------------+---------------+
| 7ce392a3c1624cd2be4b5276e8825059 | tasks.RAGFLOW |
+----------------------------------+---------------+

# Consume a task from the message queue
RAGFlow(admin)> mq pull
+------+-----+----------------+
| ack  | id  | type           |
+------+-----+----------------+
| true | cdd | ingestion_test |
+------+-----+----------------+

# User mode
# List ingestion tasks, followed by dataset id
RAGFlow(user)> list ingestion tasks from '0abe79f9423311f1ad8d38a74640adcc';
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+
| create_date               | create_time   | dataset_id                       | document_id                      | id                               | schema | status    | update_date               | update_time   | user_id                          |
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+
| 2026-05-30T20:21:06+08:00 | 1780143666289 | 0abe79f9423311f1ad8d38a74640adcc | ffe64fae423411f1a2d938a74640adcc | 8d758cd14a8b4ba8ab505003fb52017d |        | COMPLETED | 2026-05-30T20:21:26+08:00 | 1780143686431 | 2ba4881420fa11f19e9c38a74640adcc |
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+

RAGFlow(user)> list ingestion tasks;
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+
| create_date               | create_time   | dataset_id                       | document_id                      | id                               | schema | status    | update_date               | update_time   | user_id                          |
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+
| 2026-06-02T19:02:31+08:00 | 1780398151417 | 0abe79f9423311f1ad8d38a74640adcc | ffe64fae423411f1a2d938a74640adcc | e89e20d9a25848a1b79bd9345ddbfe1d |        | COMPLETED | 2026-06-02T19:02:52+08:00 | 1780398172208 | 2ba4881420fa11f19e9c38a74640adcc |
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+

# Create an ingestion task
# First argument is document id, second argument is dataset id
RAGFlow(user)> start ingestion 'ffe64fae423411f1a2d938a74640adcc' from '0abe79f9423311f1ad8d38a74640adcc';
+----------------------------------+-------------------------------------------+
| document_id                      | result                                    |
+----------------------------------+-------------------------------------------+
| ffe64fae423411f1a2d938a74640adcc | task_id: 8d758cd14a8b4ba8ab505003fb52017d |
+----------------------------------+-------------------------------------------+

# Pause an ingestion task, first argument is ingestion id
RAGFlow(user)> stop ingestion '8d758cd14a8b4ba8ab505003fb52017d';
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+
| create_date               | create_time   | dataset_id                       | document_id                      | id                               | schema | status    | update_date               | update_time   | user_id                          |
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+
| 2026-05-30T20:21:06+08:00 | 1780143666289 | 0abe79f9423311f1ad8d38a74640adcc | ffe64fae423411f1a2d938a74640adcc | 8d758cd14a8b4ba8ab505003fb52017d |        | COMPLETED | 2026-05-30T20:21:26+08:00 | 1780143686431 | 2ba4881420fa11f19e9c38a74640adcc |
+---------------------------+---------------+----------------------------------+----------------------------------+----------------------------------+--------+-----------+---------------------------+---------------+----------------------------------+

# Delete an ingestion task
RAGFlow(api/default)> remove ingestion tasks 'f366450a27d54677aec1c7090add30f0';
+---------+----------------------------------+
| remove  | task_id                          |
+---------+----------------------------------+
| success | f366450a27d54677aec1c7090add30f0 |
+---------+----------------------------------+

```

### Type of change

- [x] New Feature (non-breaking change which adds functionality)

---------

Signed-off-by: Jin Hai <haijin.chn@gmail.com>
2026-06-12 14:56:44 +08:00

183 lines
4.8 KiB
Go

//
// Copyright 2026 The InfiniFlow Authors. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
package entity
import (
"database/sql/driver"
"encoding/json"
"time"
"gorm.io/gorm"
)
// BaseModel base model
// All time fields are nullable to match Python Peewee model (null=True)
type BaseModel struct {
CreateTime *int64 `gorm:"column:create_time;index" json:"create_time,omitempty"`
CreateDate *time.Time `gorm:"column:create_date;index" json:"create_date,omitempty"`
UpdateTime *int64 `gorm:"column:update_time;index" json:"update_time,omitempty"`
UpdateDate *time.Time `gorm:"column:update_date;index" json:"update_date,omitempty"`
}
func autoModelTime() (int64, time.Time) {
now := time.Now().Local()
return now.UnixMilli(), now.Truncate(time.Second)
}
func statementHasTimeField(tx *gorm.DB, fieldNames ...string) bool {
if tx == nil || tx.Statement == nil {
return false
}
switch dest := tx.Statement.Dest.(type) {
case map[string]interface{}:
for _, fieldName := range fieldNames {
if _, ok := dest[fieldName]; ok {
return true
}
}
case []map[string]interface{}:
for _, item := range dest {
for _, fieldName := range fieldNames {
if _, ok := item[fieldName]; ok {
return true
}
}
}
}
return false
}
// BeforeCreate injects timestamps for models embedding BaseModel.
func (m *BaseModel) BeforeCreate(tx *gorm.DB) error {
timestamp, dateTime := autoModelTime()
if m.CreateTime == nil {
m.CreateTime = &timestamp
}
if m.CreateDate == nil {
m.CreateDate = &dateTime
}
if m.UpdateTime == nil {
m.UpdateTime = &timestamp
}
if m.UpdateDate == nil {
m.UpdateDate = &dateTime
}
if tx != nil && tx.Statement != nil {
if !statementHasTimeField(tx, "create_time", "CreateTime") && m.CreateTime != nil {
tx.Statement.SetColumn("CreateTime", *m.CreateTime)
}
if !statementHasTimeField(tx, "create_date", "CreateDate") && m.CreateDate != nil {
tx.Statement.SetColumn("CreateDate", *m.CreateDate)
}
if !statementHasTimeField(tx, "update_time", "UpdateTime") && m.UpdateTime != nil {
tx.Statement.SetColumn("UpdateTime", *m.UpdateTime)
}
if !statementHasTimeField(tx, "update_date", "UpdateDate") && m.UpdateDate != nil {
tx.Statement.SetColumn("UpdateDate", *m.UpdateDate)
}
}
return nil
}
// BeforeUpdate injects update timestamps for models embedding BaseModel.
func (m *BaseModel) BeforeUpdate(tx *gorm.DB) error {
timestamp, dateTime := autoModelTime()
if !statementHasTimeField(tx, "update_time", "UpdateTime") {
m.UpdateTime = &timestamp
}
if !statementHasTimeField(tx, "update_date", "UpdateDate") {
m.UpdateDate = &dateTime
}
if tx != nil && tx.Statement != nil {
if !statementHasTimeField(tx, "update_time", "UpdateTime") && m.UpdateTime != nil {
tx.Statement.SetColumn("UpdateTime", *m.UpdateTime)
}
if !statementHasTimeField(tx, "update_date", "UpdateDate") && m.UpdateDate != nil {
tx.Statement.SetColumn("UpdateDate", *m.UpdateDate)
}
}
return nil
}
func (m *BaseModel) UpdateCreateDateAndTime() error {
timestamp, dateTime := autoModelTime()
m.CreateTime = &timestamp
m.UpdateDate = &dateTime
return nil
}
func (m *BaseModel) UpdateUpdateDateAndTime() error {
timestamp, dateTime := autoModelTime()
m.UpdateTime = &timestamp
m.UpdateDate = &dateTime
return nil
}
// JSONMap is a map type that can store JSON data
type JSONMap map[string]interface{}
// Value implements driver.Valuer interface
func (j JSONMap) Value() (driver.Value, error) {
if j == nil {
return nil, nil
}
return json.Marshal(j)
}
// Scan implements sql.Scanner interface
func (j *JSONMap) Scan(value interface{}) error {
if value == nil {
*j = nil
return nil
}
b, ok := value.([]byte)
if !ok {
return json.Unmarshal([]byte(value.(string)), j)
}
return json.Unmarshal(b, j)
}
// JSONSlice is a slice type that can store JSON array data
type JSONSlice []interface{}
// Value implements driver.Valuer interface
func (j JSONSlice) Value() (driver.Value, error) {
if j == nil {
return nil, nil
}
return json.Marshal(j)
}
// Scan implements sql.Scanner interface
func (j *JSONSlice) Scan(value interface{}) error {
if value == nil {
*j = nil
return nil
}
b, ok := value.([]byte)
if !ok {
return json.Unmarshal([]byte(value.(string)), j)
}
return json.Unmarshal(b, j)
}