init

xiangj-adapter/internal/service/data_service.go

@@ -0,0 +1,79 @@
+package service
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"regexp"
+)
+
+var tableNameRegex = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)
+
+type DataService struct {
+	db     *sql.DB
+	schema string
+}
+
+func NewDataService(db *sql.DB, schema string) *DataService {
+	return &DataService{db: db, schema: schema}
+}
+
+func (s *DataService) Select(ctx context.Context, table string, offset, count int) ([]map[string]any, error) {
+	if !tableNameRegex.MatchString(table) {
+		return nil, fmt.Errorf("invalid table name")
+	}
+	if s.schema != "" && !tableNameRegex.MatchString(s.schema) {
+		return nil, fmt.Errorf("invalid schema name")
+	}
+
+	qualifiedTable := table
+	if s.schema != "" {
+		qualifiedTable = fmt.Sprintf("%s.%s", s.schema, table)
+	}
+
+	query := fmt.Sprintf("SELECT * FROM %s LIMIT $2 OFFSET $1", qualifiedTable)
+	rows, err := s.db.QueryContext(ctx, query, offset, count)
+	if err != nil {
+		return nil, fmt.Errorf("query table %s: %w", table, err)
+	}
+	defer rows.Close()
+
+	return scanRows(rows)
+}
+
+func scanRows(rows *sql.Rows) ([]map[string]any, error) {
+	columns, err := rows.Columns()
+	if err != nil {
+		return nil, fmt.Errorf("read columns: %w", err)
+	}
+
+	results := make([]map[string]any, 0)
+	values := make([]any, len(columns))
+	valuePtrs := make([]any, len(columns))
+	for i := range values {
+		valuePtrs[i] = &values[i]
+	}
+
+	for rows.Next() {
+		if err := rows.Scan(valuePtrs...); err != nil {
+			return nil, fmt.Errorf("scan row: %w", err)
+		}
+
+		row := make(map[string]any, len(columns))
+		for i, col := range columns {
+			val := values[i]
+			if b, ok := val.([]byte); ok {
+				row[col] = string(b)
+				continue
+			}
+			row[col] = val
+		}
+		results = append(results, row)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterate rows: %w", err)
+	}
+
+	return results, nil
+}