SQLServerNIO

.package(url: "https://github.com/tashda/echo-sqlserver.git", from: "0.1.0")

.target(
    name: "MyApp",
    dependencies: [
        .product(name: "SQLServerKit", package: "echo-sqlserver")
    ]
)

import SQLServerKit

let configuration = SQLServerClient.Configuration(
    hostname: "localhost",
    port: 1433,
    login: .init(
        database: "master",
        authentication: .sqlPassword(username: "sa", password: "<your_password>")
    ),
    tlsConfiguration: .makeClientConfiguration()
)

let client = try await SQLServerClient.connect(configuration: configuration)

// Simple query
let rows = try await client.query("SELECT @@VERSION AS version")
print(rows.first?.column("version")?.string ?? "<unknown>")

// Server product version (e.g. 16.0.1000.5)
let productVersion = try await client.serverVersion()
print("Server version: \(productVersion)")

// Execute with results
let result = try await client.execute("SELECT COUNT(*) as count FROM sys.tables")
print("Table count: \(result.rows.first?.column("count")?.int ?? 0)")

try await client.shutdownGracefully()

let connection = try await SQLServerConnection.connect(configuration: configuration.connection)
let databases = try await connection.listDatabases()
print("Found \(databases.count) databases")
try await connection.close()

var configuration = makeSQLServerClientConfiguration()
configuration.connection.sessionOptions = .init(
    nocount: false,
    fmtOnlyOff: true,
    additionalStatements: ["SET DEADLOCK_PRIORITY LOW;"]
)
configuration.connection.transparentNetworkIPResolution = true // try all DNS answers before failing

// Query returning rows
let rows = try await client.query("SELECT name FROM sys.databases ORDER BY name")
for row in rows {
    print(row.column("name")?.string ?? "<unknown>")
}

// Execute statements (INSERT, UPDATE, DELETE, DDL)
let result = try await client.execute("INSERT INTO Logs(Message) VALUES (N'hello')")
print("Affected rows: \(result.done.first?.rowCount ?? 0)")

// Scalar queries
let databaseName: String? = try await client.queryScalar("SELECT DB_NAME()", as: String.self)
let tableCount: Int? = try await client.queryScalar("SELECT COUNT(*) FROM sys.tables", as: Int.self)

let result = try await client.execute("EXEC dbo.usp_DoThing @Input = 1, @Out = @o OUTPUT").get()
for rv in result.returnValues {
    print("name=\(rv.name) int=\(rv.int as Any) string=\(rv.string as Any)")
}

let ss = connection.lastSessionStatePayload
let dc = connection.lastDataClassificationPayload

client.query("SELECT COUNT(*) AS count FROM sys.tables").whenSuccess { rows in
    print(rows.first?.column("count")?.int ?? 0)
}

// Use a specific connection for multiple operations
try await client.withConnection { connection in
    try await connection.changeDatabase("AdventureWorks")
    let tables = try await connection.listTables(schema: "dbo")
    return tables.count
}

// Connection state is automatically managed
print(connection.currentDatabase) // Shows current database

// Execute multiple statements as separate batches
let statements = [
    "CREATE TABLE TestTable (id INT PRIMARY KEY, name NVARCHAR(50))",
    "INSERT INTO TestTable VALUES (1, N'Test')",
    "SELECT * FROM TestTable"
]
let results = try await client.executeSeparateBatches(statements)
print("Executed \(results.count) batches")

// Execute SQL script with GO separators
let script = """
CREATE TABLE Users (id INT PRIMARY KEY, name NVARCHAR(100))
GO
INSERT INTO Users VALUES (1, N'John Doe')
GO
SELECT COUNT(*) FROM Users
GO
"""
let scriptResults = try await client.executeScript(script)

try await client.withConnection { connection in
    for try await event in connection.streamQuery("SELECT * FROM LargeTable") {
        switch event {
        case .metadata(let columns):
            print("Columns: \(columns.map(\.name))")
        case .row(let row):
            // Process each row as it arrives
            processRow(row)
        case .done(let done):
            print("Batch complete, rows: \(done.rowCount)")
        case .message(let message):
            print("Server message: \(message.message)")
        }
    }
}

import SQLServerKit

let options = SqlServerExecutionOptions(
    mode: .auto,                 // or .simple, .cursor
    rowsetFetchSize: nil,        // reserved for future rowset/cursor support
    progressThrottleMs: 120      // reserved for future package-level progress cadence
)

try await client.withConnection { connection in
    for try await event in connection.streamQuery("SELECT TOP 10000 * FROM dbo.Fixture", options: options) {
        switch event {
        case .metadata(let cols):
            // ...
            break
        case .row(let row):
            // ...
            break
        case .done(let done):
            // ...
            break
        case .message(let msg):
            // ...
            break
        }
    }
}

let monitor = SQLServerActivityMonitor(client: client)

// One-shot snapshot (no text/plan by default)
let snap = try await monitor.snapshot()
print("Processes: \(snap.processes.count), waits: \(snap.waits.count)")

// Stream snapshots every 5 seconds (default)
for try await s in monitor.streamSnapshots(every: 5.0, options: .init(includeSqlText: false, includeQueryPlan: false)) {
    print("Top waits delta: \(s.waitsDelta?.first?.waitType ?? "<none>")")
}

// Include query text and/or plans (heavy; opt-in)
let rich = try await monitor.snapshot(options: .init(includeSqlText: true, includeQueryPlan: false))
print(rich.processes.first?.request?.sqlText ?? "<no text>")

// Management action: kill a session (requires privileges)
try await monitor.killSession(sessionId: 55)

let dbSec = SQLServerSecurityClient(client: client)

// GRANT on database (no ON clause)
try await dbSec.grant(permission: .connect, on: .database(nil), to: "public")

// GRANT on schema
try await dbSec.grant(permission: .alterAnySchema, on: .schema("dbo"), to: "dbo")

// GRANT column-level SELECT on object
let oid = ObjectIdentifier(schema: "dbo", name: "Employees", kind: .table)
try await dbSec.grant(permission: .select, on: .column(oid, ["id", "name"]), to: "public")

// Detailed permission listing including DB- and schema-scope
let perms = try await dbSec.listPermissionsDetailed(principal: "public")

// Application roles
try await dbSec.createApplicationRole(name: "AppRole", password: "p@ss", defaultSchema: "dbo")
try await dbSec.alterApplicationRole(name: "AppRole", newName: "AppRole2")
try await dbSec.dropApplicationRole(name: "AppRole2")

// Schema helpers
try await dbSec.createSchema(name: "ops", authorization: "dbo")
try await dbSec.transferObjectToSchema(objectSchema: "dbo", objectName: "SomeTable", newSchema: "ops")
try await dbSec.alterAuthorizationOnSchema(schema: "ops", principal: "dbo")
try await dbSec.dropSchema(name: "ops")

let srvSec = SQLServerServerSecurityClient(client: client)
try await srvSec.createCertificateLogin(name: "cert_login", certificateName: "MyCert")
try await srvSec.createAsymmetricKeyLogin(name: "asym_login", asymmetricKeyName: "MyKey")
try await srvSec.createExternalLogin(name: "aad_login")

let serverSec = SQLServerServerSecurityClient(client: client)

// List logins
let logins = try await serverSec.listLogins()

// Create a SQL login (CHECK_POLICY/EXPIRATION and defaults supported)
try await serverSec.createSqlLogin(
    name: "app_login",
    password: "<your_password>",
    options: .init(defaultDatabase: "MyDb", checkPolicy: true, checkExpiration: false)
)

// Add to a server role
try await serverSec.addMemberToServerRole(role: "securityadmin", principal: "app_login")

// Grant server permission
try await serverSec.grant(permission: .viewServerState, to: "app_login")

// Credentials
try await serverSec.createCredential(name: "s3_cred", identity: "DOMAIN\\svc_user", secret: "secret")
try await serverSec.alterCredential(name: "s3_cred", identity: nil, secret: "newSecret")
try await serverSec.dropCredential(name: "s3_cred")

try await client.withConnection { connection in
    // Begin transaction
    _ = try await connection.execute("BEGIN TRANSACTION")
    
    do {
        // Your transactional operations
        _ = try await connection.execute("INSERT INTO Orders (customer_id) VALUES (123)")
        _ = try await connection.execute("UPDATE Inventory SET quantity = quantity - 1 WHERE product_id = 456")
        
        // Commit if all operations succeed
        _ = try await connection.execute("COMMIT")
    } catch {
        // Rollback on any error
        _ = try await connection.execute("ROLLBACK")
        throw error
    }
}

// Tables and columns with comments (opt-in)
let tables = try await client.listTables(schema: "dbo", includeComments: true)
let columns = try await client.listColumns(schema: "dbo", table: "Employees", includeComments: true)
for column in columns {
    print("\(column.name) \(column.typeName) comment=\(column.comment ?? "<none>")")
}

// Procedure / function parameters
let parameters = try await client.listParameters(schema: "dbo", object: "usp_UpdateEmployee")
let inputs = parameters.filter { !$0.isReturnValue }

// Keys & indexes
let primaryKeys = try await client.listPrimaryKeys(schema: "dbo", table: "Employees")
let indexes = try await client.listIndexes(schema: "dbo", table: "EmployeeAudit")

// Foreign keys & dependencies
let foreignKeys = try await client.listForeignKeys(schema: "dbo", table: "EmployeeAudit")
let dependencies = try await client.listDependencies(schema: "dbo", object: "Employees")

// Routines & triggers (comments opt-in)
let procedures = try await client.listProcedures(schema: "dbo", includeComments: true)
let functions = try await client.listFunctions(schema: "dbo", includeComments: true)
let triggers = try await client.listTriggers(schema: "dbo", table: "EmployeeAudit", includeComments: true)

var clientConfig = makeSQLServerClientConfiguration()
clientConfig.connection.metadataConfiguration.includeRoutineDefinitions = true
let client = try await SQLServerClient.connect(configuration: clientConfig)

let procedures = try await client.listProcedures(schema: "dbo") // definitions populated

let identifiers = [
    SQLServerMetadataObjectIdentifier(database: nil, schema: "dbo", name: "usp_ReportSales", kind: .procedure),
    SQLServerMetadataObjectIdentifier(database: nil, schema: "dbo", name: "vw_Sales", kind: .view)
]

let definitions = try await client.fetchObjectDefinitions(identifiers)

let single = try await client.fetchObjectDefinition(schema: "dbo", name: "usp_ReportSales", kind: .procedure)
print(single?.definition)

let hits = try await client.searchMetadata(
    query: "Customer",
    scopes: [.objectNames, .definitions, .columns]
)

for hit in hits {
    print("\(hit.matchKind): \(hit.schema).\(hit.name) -> \(hit.detail ?? "")")
}

let admin = SQLServerAdministrationClient(client: client)
let logins = try await admin.listServerLogins()

// Create tables with comprehensive options
let columns = [
    SQLServerColumnDefinition(name: "id", definition: .standard(.init(dataType: .int, isPrimaryKey: true))),
    SQLServerColumnDefinition(name: "name", definition: .standard(.init(dataType: .nvarchar(length: .length(100)), comment: "User name")))
]
try await admin.createTable(name: "Users", columns: columns)

let routineClient = SQLServerRoutineClient(client: client)

// Create stored procedure
let parameters = [
    ProcedureParameter(name: "user_id", dataType: .int),
    ProcedureParameter(name: "result", dataType: .nvarchar(length: .length(100)), direction: .output)
]
let body = """
BEGIN
    SELECT @result = name FROM Users WHERE id = @user_id
END
"""
try await routineClient.createStoredProcedure(name: "GetUserName", parameters: parameters, body: body)

// Create scalar function
try await routineClient.createFunction(
    name: "CalculateAge",
    parameters: [FunctionParameter(name: "birth_date", dataType: .date)],
    returnType: .int,
    body: "BEGIN RETURN DATEDIFF(YEAR, @birth_date, GETDATE()) END"
)

// Execute-as / impersonation options
let executeAsOptions = RoutineOptions(executeAs: "dbo", withRecompile: true)
try await routineClient.createStoredProcedure(name: "GetAdminData", body: body, options: executeAsOptions)

// Create table-valued function
let tableDefinition = [
    TableValuedFunctionColumn(name: "id", dataType: .int),
    TableValuedFunctionColumn(name: "name", dataType: .nvarchar(length: .length(100)))
]
try await routineClient.createTableValuedFunction(
    name: "GetActiveUsers",
    tableDefinition: tableDefinition,
    body: """
    SELECT id, name
    FROM Users
    WHERE active = 1
    """
)
// Bodies that start with SELECT/WITH automatically become INSERT ... RETURN blocks.
// Provide your own BEGIN/INSERT statements if you need side effects or multiple phases.

let viewClient = SQLServerViewClient(client: client)

// Create view
try await viewClient.createView(
    name: "ActiveUsers",
    query: "SELECT id, name, email FROM Users WHERE active = 1"
)

// Create indexed view (materialized view)
try await viewClient.createIndexedView(
    name: "UserSummary",
    query: "SELECT department, COUNT_BIG(*) as user_count FROM dbo.Users GROUP BY department",
    indexName: "IX_UserSummary_department",
    indexColumns: ["department"]
)

let indexClient = SQLServerIndexClient(client: client)

// Create nonclustered index
try await indexClient.createIndex(
    name: "IX_Users_Email",
    table: "Users",
    columns: [IndexColumn(name: "email")]
)

// Create unique index
try await indexClient.createUniqueIndex(
    name: "IX_Users_Username_Unique",
    table: "Users",
    columns: [IndexColumn(name: "username")]
)

// Create index with included columns
try await indexClient.createIndex(
    name: "IX_Users_LastName_Incl",
    table: "Users",
    columns: [
        IndexColumn(name: "last_name"),
        IndexColumn(name: "first_name", isIncluded: true),
        IndexColumn(name: "email", isIncluded: true)
    ]
)

// Rebuild index
try await indexClient.rebuildIndex(name: "IX_Users_Email", table: "Users")

let bulk = SQLServerBulkClient(client: client)
var options = SQLServerBulkCopyOptions(
    table: "Costs",
    columns: ["category", "amount", "booked_at"],
    batchSize: 5_000
)
options.tableLock = true          // faster, blocks other writers during each batch
options.fireTriggers = false      // defaults: constraints checked, triggers fired, NULLs kept
let rows = [
    SQLServerBulkCopyRow(values: [.nString("Hardware"), .string("123.45"), .string("2024-02-29 13:14:15")]),
    SQLServerBulkCopyRow(values: [.nString("Software"), .decimal("300.00"), .null])
]
let summary = try await bulk.copy(rows: rows, options: options)
print("Copied \(summary.totalRows) rows in \(summary.batchesExecuted) batches (\(summary.method))")

let parameter = SQLServerTableValuedParameter(
    name: "NewUsers",
    columns: [
        .init(name: "id", dataType: .int),
        .init(name: "name", dataType: .nvarchar(length: .length(100)))
    ],
    rows: [
        .init(values: [.int(1), .nString("Alice")]),
        .init(values: [.int(2), .nString("Bob")])
    ]
)

try await client.execute("""
INSERT INTO dbo.Users (id, name)
SELECT id, name FROM @NewUsers;
""", tableParameters: [parameter])

let constraintClient = SQLServerConstraintClient(client: client)

// Add foreign key constraint
try await constraintClient.addForeignKey(
    name: "FK_Orders_Users",
    table: "Orders",
    columns: ["user_id"],
    referencedTable: "Users",
    referencedColumns: ["id"],
    options: ForeignKeyOptions(onDelete: .cascade)
)

// Add check constraint
try await constraintClient.addCheckConstraint(
    name: "CK_Users_Age",
    table: "Users",
    expression: "age >= 0 AND age <= 150"
)

// Add unique constraint
try await constraintClient.addUniqueConstraint(
    name: "UQ_Users_Email",
    table: "Users",
    columns: ["email"]
)

let triggerClient = SQLServerTriggerClient(client: client)

// Create audit trigger
let auditBody = """
BEGIN
    SET NOCOUNT ON;
    INSERT INTO UserAudit (user_id, operation, audit_date)
    SELECT inserted.id, 'INSERT', GETDATE()
    FROM inserted;
END
"""
try await triggerClient.createTrigger(
    name: "tr_Users_Audit",
    table: "Users",
    timing: .after,
    events: [.insert, .update, .delete],
    body: auditBody
)

// Enable/disable triggers
try await triggerClient.disableTrigger(name: "tr_Users_Audit", table: "Users")
try await triggerClient.enableTrigger(name: "tr_Users_Audit", table: "Users")

// Database scope (backward compatible alias SQLServerDatabaseSecurityClient)
let securityClient = SQLServerSecurityClient(client: client)

// Create database user
try await securityClient.createUser(name: "app_user", options: UserOptions(defaultSchema: "dbo"))

// Create custom role
try await securityClient.createRole(name: "data_analysts")

// Add user to role
try await securityClient.addUserToRole(user: "app_user", role: "data_analysts")

// Grant permissions
try await securityClient.grantPermission(permission: .select, on: "Users", to: "data_analysts")
try await securityClient.grantPermission(permission: .insert, on: "Users", to: "data_analysts")

// Add user to built-in database roles
try await securityClient.addUserToDatabaseRole(user: "app_user", role: .dbDataReader)

// List permissions
let permissions = try await securityClient.listPermissions(principal: "app_user")

// Agent status (running + XPs enabled)
let metadata = SQLServerMetadataClient(connection: try await client.withConnection { $0 })
let status = try await metadata.fetchAgentStatus()
print("Agent running=\(status.isSqlAgentRunning) enabled=\(status.isSqlAgentEnabled)")

// Optional: preflight environment and fail fast with guidance
let agent = SQLServerAgentClient(client: client)
try await agent.preflightAgentEnvironment().get() // set requireProxyPrereqs: true to validate proxy/credential perms

// List jobs and start one
let jobs = try await agent.listJobs()
if let job = jobs.first, status.isSqlAgentRunning, status.isSqlAgentEnabled {
    try await agent.startJob(named: job.name)
}

// Create a job, add a T-SQL step, attach to server, start/stop, enable/disable, fetch history
let jobName = "nio_sample_job"
try await agent.createJob(named: jobName, description: "Created from SQLServerNIO", enabled: true)
try await agent.addTSQLStep(jobName: jobName, stepName: "main", command: "SELECT 1;", database: "master")
// createJob automatically associates the job with the local server.
// addJobServer is safe to call again; it is effectively idempotent.
try await agent.addJobServer(jobName: jobName)
try await agent.startJob(named: jobName)
let running = try await agent.listRunningJobs()
print("Running jobs: \(running.map(\.name))")
let history = try await agent.listJobHistory(jobName: jobName, top: 10)
try await agent.stopJob(named: jobName) // no-op if already completed
try await agent.enableJob(named: jobName, enabled: false)
try await agent.deleteJob(named: jobName)

// Schedules, operators, alerts
// Create a daily schedule and attach to a job
try await agent.createSchedule(named: "nio_daily", enabled: true, freqType: 4, freqInterval: 1, activeStartTime: 0)
try await agent.attachSchedule(scheduleName: "nio_daily", toJob: jobName)
let schedules = try await agent.listSchedules(forJob: jobName)

// Operators and alerts
try await agent.createOperator(name: "nio_operator", emailAddress: "devnull@example.com")
try await agent.createAlert(name: "nio_severity1", severity: 1)
try await agent.addNotification(alertName: "nio_severity1", operatorName: "nio_operator", method: 1)
let operators = try await agent.listOperators()
let alerts = try await agent.listAlerts()

### Extended Agent Features

// Job categories and rename/owner
try await agent.createCategory(name: "nio_category")
try await agent.setJobCategory(named: jobName, categoryName: "nio_category")
try await agent.renameJob(named: jobName, to: "nio_job_renamed")
let currentLogin = try await client.queryScalar("SELECT SUSER_SNAME()", as: String.self) ?? "sa"
try await agent.changeJobOwner(named: "nio_job_renamed", ownerLoginName: currentLogin)

// Email notifications for jobs (requires operator, Database Mail for delivery)
try await agent.createOperator(name: "nio_operator", emailAddress: "devnull@example.com")
try await agent.setJobEmailNotification(jobName: "nio_job_renamed", operatorName: "nio_operator", notifyLevel: 2) // on failure

// Schedules: create, attach, list next run times
try await agent.createSchedule(named: "nio_daily", enabled: true, freqType: 4, freqInterval: 1, activeStartTime: 0)
try await agent.attachSchedule(scheduleName: "nio_daily", toJob: "nio_job_renamed")
let nextRuns = try await agent.listJobNextRunTimes(jobName: "nio_job_renamed")

// Generic steps for other subsystems (CmdExec, PowerShell, SSIS). Include optional proxy usage
try await agent.addStep(jobName: "nio_job_renamed", stepName: "cmd", subsystem: "TSQL", command: "SELECT 1;", database: "master")

// Step flow control and retry/output options
try await agent.configureStep(jobName: "nio_job_renamed", stepName: "cmd", onSuccessAction: 3, onFailAction: 2, retryAttempts: 1, retryIntervalMinutes: 1, outputFileName: "C:\\temp\\nio_job_cmd.txt", appendOutputFile: true)

// Start job from a specific step name (via stored procedure)
_ = try await client.execute("EXEC msdb.dbo.sp_start_job @job_name = N'nio_job_renamed', @step_name = N'cmd';")

// Proxies and credentials (require elevated permissions)
try await agent.createCredential(name: "nio_cred", identity: "id_nio", secret: "s3cr3t!")
try await agent.createProxy(name: "nio_proxy", credentialName: "nio_cred", description: "for CmdExec")
try await agent.grantLoginToProxy(proxyName: "nio_proxy", loginName: currentLogin)
try await agent.grantProxyToSubsystem(proxyName: "nio_proxy", subsystem: "CmdExec")
let proxies = try await agent.listProxies()

## Error Handling & Retries


Errors surface as `SQLServerError`:

- `.authenticationFailed`
- `.connectionClosed`
- `.timeout(description:underlying:)`
- `.protocol(TDSError)`
- `.transient(Error)`
- `.unknown(Error)`

Configure retries via `SQLServerRetryConfiguration` on your connection/client configuration.

## Testing

```bash
swift test   # unit tests; integration tests skip without a server
docker run -d --name sqlserver-test -e ACCEPT_EULA=Y -e MSSQL_SA_PASSWORD='Your_password1' \
    -p 1433:1433 mcr.microsoft.com/mssql/server:2022-latest
SQLSERVER_TEST_URL='sqlserver://sa:Your_password1@localhost:1433/master?trustServerCertificate=true' swift test