313 lines
12 KiB
Go
313 lines
12 KiB
Go
package runtime
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"database/sql"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"strings"
|
|
"time"
|
|
|
|
"browser.local/run/protocol"
|
|
_ "modernc.org/sqlite"
|
|
)
|
|
|
|
const sqliteSchemaProbeDriver = "sqlite"
|
|
|
|
// SQLiteSchemaProbeExecutor performs only fixed SQLite introspection queries.
|
|
// The assignment carries no SQL and the only local target is a package-scoped,
|
|
// logical database key.
|
|
type SQLiteSchemaProbeExecutor struct{ resolver WorkspaceResolver }
|
|
|
|
func NewSQLiteSchemaProbeExecutor(workspaceRoot string) *SQLiteSchemaProbeExecutor {
|
|
return &SQLiteSchemaProbeExecutor{resolver: NewWorkspaceResolver(workspaceRoot)}
|
|
}
|
|
|
|
func (executor *SQLiteSchemaProbeExecutor) Execute(ctx context.Context, assignment protocol.RunJobAssignment) LifecycleExecutionResult {
|
|
if err := protocol.ValidateRunJobAssignment(assignment); err != nil {
|
|
return sqliteSchemaProbeFailure(assignment, "invalid_request", false)
|
|
}
|
|
request := *assignment.ExecutionInput.SQLiteSchemaProbe
|
|
probe := protocol.SQLiteSchemaProbeResult{RequestID: request.RequestID, JobID: assignment.JobID, Binding: request.Binding, Status: "failed", Limits: request.Limits}
|
|
scope, err := executor.resolver.Scope(assignment.ServerInstanceID, assignment.ExecutionInput.WorkspaceScope)
|
|
if err != nil {
|
|
return sqliteSchemaProbeTerminalFailure(probe, "target_unavailable", false)
|
|
}
|
|
path, err := executor.resolver.ExistingTarget(scope, assignment.TargetKey)
|
|
if err != nil {
|
|
return sqliteSchemaProbeTerminalFailure(probe, "target_unavailable", false)
|
|
}
|
|
sourceFingerprint, err := fingerprintSQLiteSource(path)
|
|
if err != nil {
|
|
return sqliteSchemaProbeTerminalFailure(probe, "source_unavailable", true)
|
|
}
|
|
probe.SourceFingerprint = sourceFingerprint
|
|
|
|
probeCtx, cancel := context.WithTimeout(ctx, time.Duration(request.Limits.TimeoutMS)*time.Millisecond)
|
|
defer cancel()
|
|
database, err := sql.Open(sqliteSchemaProbeDriver, "file:"+path+"?mode=ro")
|
|
if err != nil {
|
|
return sqliteSchemaProbeTerminalFailure(probe, "sqlite_open_failed", true)
|
|
}
|
|
defer database.Close()
|
|
database.SetMaxOpenConns(1)
|
|
database.SetConnMaxLifetime(time.Minute)
|
|
if _, err := database.ExecContext(probeCtx, "PRAGMA query_only = ON"); err != nil {
|
|
return sqliteSchemaProbeTerminalFailure(probe, sqliteProbeErrorCode(probeCtx, err), true)
|
|
}
|
|
objects, err := inspectSQLiteSchema(probeCtx, database, request.Limits)
|
|
if err != nil {
|
|
return sqliteSchemaProbeTerminalFailure(probe, sqliteProbeErrorCode(probeCtx, err), true)
|
|
}
|
|
probe.Objects = objects
|
|
if sourceFingerprintAfter, err := fingerprintSQLiteSource(path); err != nil || sourceFingerprintAfter != probe.SourceFingerprint {
|
|
return sqliteSchemaProbeTerminalFailure(probe, "source_changed", true)
|
|
}
|
|
probe.SchemaFingerprint = digestValue(schemaFingerprintInput(objects))
|
|
probe.ObservedAt = time.Now().UTC()
|
|
probe.Status = "succeeded"
|
|
if !finalizeSQLiteSchemaProbe(&probe) || sqliteSchemaProbeSize(probe) > request.Limits.MaxResultBytes {
|
|
return sqliteSchemaProbeTerminalFailure(probe, "result_limit_exceeded", false)
|
|
}
|
|
return LifecycleExecutionResult{State: lifecycleResultStateSucceeded, Progress: protocol.RunJobProgressReport{Percent: 100, Message: "SQLite schema probe completed"}, Message: "SQLite schema probe completed", ExecutionResult: protocol.RunJobExecutionResult{Kind: "sqlite.schema-probe", Checksum: probe.ResultDigest, SizeBytes: int64(sqliteSchemaProbeSize(probe)), Summary: "bounded query-only SQLite schema metadata", SQLiteSchemaProbe: &probe}}
|
|
}
|
|
|
|
func sqliteSchemaProbeFailure(assignment protocol.RunJobAssignment, code string, retryable bool) LifecycleExecutionResult {
|
|
probe := protocol.SQLiteSchemaProbeResult{JobID: assignment.JobID, Status: "failed", SafeError: protocol.SQLiteSchemaProbeSafeError{Code: code, Retryable: retryable}}
|
|
if assignment.ExecutionInput.SQLiteSchemaProbe != nil {
|
|
probe.RequestID, probe.Binding, probe.Limits = assignment.ExecutionInput.SQLiteSchemaProbe.RequestID, assignment.ExecutionInput.SQLiteSchemaProbe.Binding, assignment.ExecutionInput.SQLiteSchemaProbe.Limits
|
|
}
|
|
return sqliteSchemaProbeTerminalFailure(probe, code, retryable)
|
|
}
|
|
|
|
func sqliteSchemaProbeTerminalFailure(probe protocol.SQLiteSchemaProbeResult, code string, retryable bool) LifecycleExecutionResult {
|
|
probe.Status, probe.ObservedAt, probe.SafeError = "failed", time.Now().UTC(), protocol.SQLiteSchemaProbeSafeError{Code: code, Retryable: retryable}
|
|
_ = finalizeSQLiteSchemaProbe(&probe)
|
|
return LifecycleExecutionResult{State: lifecycleResultStateFailed, Progress: protocol.RunJobProgressReport{Percent: 100, Message: "SQLite schema probe failed"}, Message: "SQLite schema probe failed", ErrorCode: code, Retryable: retryable, ExecutionResult: protocol.RunJobExecutionResult{Kind: "sqlite.schema-probe", Checksum: probe.ResultDigest, SizeBytes: int64(sqliteSchemaProbeSize(probe)), Summary: "bounded query-only SQLite schema probe failed", SQLiteSchemaProbe: &probe}}
|
|
}
|
|
|
|
func inspectSQLiteSchema(ctx context.Context, database *sql.DB, limits protocol.SQLiteSchemaProbeLimits) ([]protocol.SQLiteSchemaProbeObject, error) {
|
|
rows, err := database.QueryContext(ctx, "SELECT name, type FROM sqlite_schema WHERE type IN ('table', 'view') AND name NOT LIKE 'sqlite_%' ORDER BY type, name LIMIT ?", limits.MaxObjects)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
type sqliteObjectIdentity struct{ name, kind string }
|
|
identities := make([]sqliteObjectIdentity, 0, limits.MaxObjects)
|
|
for rows.Next() {
|
|
var name, kind string
|
|
if err := rows.Scan(&name, &kind); err != nil {
|
|
rows.Close()
|
|
return nil, err
|
|
}
|
|
identities = append(identities, sqliteObjectIdentity{name: name, kind: kind})
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return nil, err
|
|
}
|
|
objects := make([]protocol.SQLiteSchemaProbeObject, 0, len(identities))
|
|
for _, identity := range identities {
|
|
object, err := inspectSQLiteObject(ctx, database, identity.name, identity.kind, limits, len(objects) < limits.MaxCardinalityReads)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
objects = append(objects, object)
|
|
}
|
|
return objects, nil
|
|
}
|
|
|
|
func inspectSQLiteObject(ctx context.Context, database *sql.DB, name, kind string, limits protocol.SQLiteSchemaProbeLimits, includeCardinality bool) (protocol.SQLiteSchemaProbeObject, error) {
|
|
object := protocol.SQLiteSchemaProbeObject{ObjectHash: digestValue(kind + "\x00" + name), Kind: kind, NameFingerprint: digestValue(name)}
|
|
columns, err := database.QueryContext(ctx, "SELECT cid, name, type, [notnull], pk FROM pragma_table_info(?) ORDER BY cid LIMIT ?", name, limits.MaxColumnsPerObject)
|
|
if err != nil {
|
|
return object, err
|
|
}
|
|
for columns.Next() {
|
|
var ordinal, notNull, primaryKey int
|
|
var columnName, declaredType string
|
|
if err := columns.Scan(&ordinal, &columnName, &declaredType, ¬Null, &primaryKey); err != nil {
|
|
columns.Close()
|
|
return object, err
|
|
}
|
|
nullable := notNull == 0
|
|
object.DeclaredColumns = append(object.DeclaredColumns, protocol.SQLiteSchemaProbeColumn{NameFingerprint: digestValue(columnName), DeclaredType: safeSQLiteDeclaredType(declaredType), Nullable: &nullable, PrimaryKey: primaryKey != 0, Ordinal: ordinal})
|
|
}
|
|
if err := columns.Close(); err != nil {
|
|
return object, err
|
|
}
|
|
indexes, err := database.QueryContext(ctx, "SELECT name, [unique] FROM pragma_index_list(?) ORDER BY seq LIMIT ?", name, limits.MaxIndexesPerObject)
|
|
if err != nil {
|
|
return object, err
|
|
}
|
|
type sqliteIndexIdentity struct {
|
|
name string
|
|
unique bool
|
|
}
|
|
indexIdentities := make([]sqliteIndexIdentity, 0, limits.MaxIndexesPerObject)
|
|
for indexes.Next() {
|
|
var indexName string
|
|
var unique int
|
|
if err := indexes.Scan(&indexName, &unique); err != nil {
|
|
indexes.Close()
|
|
return object, err
|
|
}
|
|
indexIdentities = append(indexIdentities, sqliteIndexIdentity{name: indexName, unique: unique != 0})
|
|
}
|
|
if err := indexes.Close(); err != nil {
|
|
return object, err
|
|
}
|
|
for _, identity := range indexIdentities {
|
|
item, err := inspectSQLiteIndex(ctx, database, identity.name, identity.unique, limits.MaxColumnsPerObject)
|
|
if err != nil {
|
|
return object, err
|
|
}
|
|
object.Indexes = append(object.Indexes, item)
|
|
}
|
|
foreignKeys, err := database.QueryContext(ctx, "SELECT [table], [from], [to] FROM pragma_foreign_key_list(?) ORDER BY id, seq LIMIT ?", name, limits.MaxForeignKeys)
|
|
if err != nil {
|
|
return object, err
|
|
}
|
|
for foreignKeys.Next() {
|
|
var destination, from, to string
|
|
if err := foreignKeys.Scan(&destination, &from, &to); err != nil {
|
|
foreignKeys.Close()
|
|
return object, err
|
|
}
|
|
object.ForeignKeys = append(object.ForeignKeys, protocol.SQLiteSchemaProbeForeignKey{FromColumnHash: digestValue(from), ToObjectHash: digestValue("table\x00" + destination), ToColumnHash: digestValue(to)})
|
|
}
|
|
if err := foreignKeys.Close(); err != nil {
|
|
return object, err
|
|
}
|
|
if includeCardinality {
|
|
var count int64
|
|
if err := database.QueryRowContext(ctx, "SELECT count(*) FROM "+quoteSQLiteIdentifier(name)).Scan(&count); err != nil {
|
|
return object, err
|
|
}
|
|
object.ApproximateRows = &count
|
|
}
|
|
if limits.MaxSampleRows > 0 {
|
|
rows, err := database.QueryContext(ctx, "SELECT * FROM "+quoteSQLiteIdentifier(name)+" LIMIT ?", limits.MaxSampleRows)
|
|
if err != nil {
|
|
return object, err
|
|
}
|
|
columns, err := rows.Columns()
|
|
if err != nil {
|
|
rows.Close()
|
|
return object, err
|
|
}
|
|
for rows.Next() {
|
|
values := make([]any, len(columns))
|
|
pointers := make([]any, len(columns))
|
|
for i := range values {
|
|
pointers[i] = &values[i]
|
|
}
|
|
if err := rows.Scan(pointers...); err != nil {
|
|
rows.Close()
|
|
return object, err
|
|
}
|
|
object.SampleFingerprints = append(object.SampleFingerprints, digestValue(canonicalSQLiteRow(columns, values)))
|
|
}
|
|
if err := rows.Close(); err != nil {
|
|
return object, err
|
|
}
|
|
}
|
|
return object, nil
|
|
}
|
|
|
|
func inspectSQLiteIndex(ctx context.Context, database *sql.DB, name string, unique bool, maxColumns int) (protocol.SQLiteSchemaProbeIndex, error) {
|
|
index := protocol.SQLiteSchemaProbeIndex{NameFingerprint: digestValue(name), Unique: unique}
|
|
rows, err := database.QueryContext(ctx, "SELECT name FROM pragma_index_info(?) ORDER BY seqno LIMIT ?", name, maxColumns)
|
|
if err != nil {
|
|
return index, err
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var column string
|
|
if err := rows.Scan(&column); err != nil {
|
|
return index, err
|
|
}
|
|
index.ColumnHashes = append(index.ColumnHashes, digestValue(column))
|
|
}
|
|
return index, rows.Err()
|
|
}
|
|
|
|
func quoteSQLiteIdentifier(value string) string {
|
|
return `"` + strings.ReplaceAll(value, `"`, `""`) + `"`
|
|
}
|
|
func digestBytes(value []byte) string {
|
|
sum := sha256.Sum256(value)
|
|
return "sha256:" + hex.EncodeToString(sum[:])
|
|
}
|
|
func digestValue(value string) string { return digestBytes([]byte(value)) }
|
|
func fingerprintSQLiteSource(path string) (string, error) {
|
|
file, err := os.Open(path)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer file.Close()
|
|
hash := sha256.New()
|
|
if _, err := io.Copy(hash, file); err != nil {
|
|
return "", err
|
|
}
|
|
return "sha256:" + hex.EncodeToString(hash.Sum(nil)), nil
|
|
}
|
|
func schemaFingerprintInput(objects []protocol.SQLiteSchemaProbeObject) string {
|
|
body, _ := json.Marshal(objects)
|
|
return string(body)
|
|
}
|
|
func finalizeSQLiteSchemaProbe(probe *protocol.SQLiteSchemaProbeResult) bool {
|
|
probe.ResultDigest = ""
|
|
body, err := json.Marshal(probe)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
probe.ResultDigest = digestBytes(body)
|
|
return true
|
|
}
|
|
func sqliteSchemaProbeSize(probe protocol.SQLiteSchemaProbeResult) int {
|
|
body, _ := json.Marshal(probe)
|
|
return len(body)
|
|
}
|
|
func canonicalSQLiteRow(columns []string, values []any) string {
|
|
body, _ := json.Marshal(struct {
|
|
Columns []string `json:"columns"`
|
|
Values []any `json:"values"`
|
|
}{columns, values})
|
|
return string(body)
|
|
}
|
|
|
|
func safeSQLiteDeclaredType(value string) string {
|
|
value = strings.ToUpper(strings.TrimSpace(value))
|
|
if value == "" {
|
|
return ""
|
|
}
|
|
if len(value) > 80 {
|
|
return "OTHER"
|
|
}
|
|
for _, char := range value {
|
|
if (char >= 'A' && char <= 'Z') || (char >= '0' && char <= '9') || strings.ContainsRune("_(), ", char) {
|
|
continue
|
|
}
|
|
return "OTHER"
|
|
}
|
|
return value
|
|
}
|
|
|
|
func sqliteProbeErrorCode(ctx context.Context, err error) string {
|
|
if errors.Is(ctx.Err(), context.Canceled) {
|
|
return "cancelled"
|
|
}
|
|
if errors.Is(ctx.Err(), context.DeadlineExceeded) {
|
|
return "timeout"
|
|
}
|
|
lower := strings.ToLower(fmt.Sprint(err))
|
|
if strings.Contains(lower, "locked") || strings.Contains(lower, "busy") {
|
|
return "database_busy"
|
|
}
|
|
return "sqlite_read_failed"
|
|
}
|