synor/sdk/flutter/lib/src/job.dart

/// Job tracking for Synor Compute SDK
library synor_compute.job;

import 'dart:async';
import 'dart:convert';

import 'package:web_socket_channel/web_socket_channel.dart';

import 'types.dart';

/// Result of a compute job
class JobResult<T> {
  /// Unique job identifier
  final String jobId;

  /// Current job status
  final JobStatus status;

  /// Result data (if completed)
  final T? result;

  /// Error message (if failed)
  final String? error;

  /// Execution time in milliseconds
  final int? executionTimeMs;

  /// Cost in credits
  final double? cost;

  /// Processor that executed the job
  final ProcessorType? processor;

  /// Metadata from execution
  final Map<String, dynamic>? metadata;

  const JobResult({
    required this.jobId,
    required this.status,
    this.result,
    this.error,
    this.executionTimeMs,
    this.cost,
    this.processor,
    this.metadata,
  });

  /// Whether the job completed successfully
  bool get isSuccess => status == JobStatus.completed && result != null;

  /// Whether the job failed
  bool get isFailed => status == JobStatus.failed;

  /// Whether the job is still running
  bool get isRunning => !status.isTerminal;

  factory JobResult.fromJson(
    Map<String, dynamic> json,
    T Function(dynamic)? resultParser,
  ) {
    final status = JobStatus.fromString(json['status'] as String);
    T? result;

    if (json['result'] != null && resultParser != null) {
      result = resultParser(json['result']);
    }

    return JobResult<T>(
      jobId: json['job_id'] as String,
      status: status,
      result: result,
      error: json['error'] as String?,
      executionTimeMs: json['execution_time_ms'] as int?,
      cost: (json['cost'] as num?)?.toDouble(),
      processor: json['processor'] != null
          ? ProcessorType.fromString(json['processor'] as String)
          : null,
      metadata: json['metadata'] as Map<String, dynamic>?,
    );
  }

  /// Transform the result to a different type
  JobResult<R> map<R>(R Function(T) transform) {
    return JobResult<R>(
      jobId: jobId,
      status: status,
      result: result != null ? transform(result as T) : null,
      error: error,
      executionTimeMs: executionTimeMs,
      cost: cost,
      processor: processor,
      metadata: metadata,
    );
  }

  @override
  String toString() {
    if (isSuccess) {
      return 'JobResult(id: $jobId, status: ${status.value}, '
          'time: ${executionTimeMs}ms, cost: \$${cost?.toStringAsFixed(6)})';
    } else if (isFailed) {
      return 'JobResult(id: $jobId, status: ${status.value}, error: $error)';
    }
    return 'JobResult(id: $jobId, status: ${status.value})';
  }
}

/// Job status update event
class JobStatusUpdate {
  final String jobId;
  final JobStatus status;
  final double? progress;
  final String? message;
  final DateTime timestamp;

  const JobStatusUpdate({
    required this.jobId,
    required this.status,
    this.progress,
    this.message,
    required this.timestamp,
  });

  factory JobStatusUpdate.fromJson(Map<String, dynamic> json) {
    return JobStatusUpdate(
      jobId: json['job_id'] as String,
      status: JobStatus.fromString(json['status'] as String),
      progress: (json['progress'] as num?)?.toDouble(),
      message: json['message'] as String?,
      timestamp: json['timestamp'] != null
          ? DateTime.parse(json['timestamp'] as String)
          : DateTime.now(),
    );
  }
}

/// Job handle for tracking and managing a submitted job
class Job<T> {
  final String jobId;
  final String _baseUrl;
  final Map<String, String> _headers;
  final T Function(dynamic)? _resultParser;

  WebSocketChannel? _wsChannel;
  StreamController<JobStatusUpdate>? _statusController;
  JobResult<T>? _cachedResult;
  bool _isDisposed = false;

  Job({
    required this.jobId,
    required String baseUrl,
    required Map<String, String> headers,
    T Function(dynamic)? resultParser,
  })  : _baseUrl = baseUrl,
        _headers = headers,
        _resultParser = resultParser;

  /// Stream of status updates for this job
  Stream<JobStatusUpdate> get statusUpdates {
    _statusController ??= StreamController<JobStatusUpdate>.broadcast(
      onListen: _startWebSocket,
      onCancel: _stopWebSocket,
    );
    return _statusController!.stream;
  }

  void _startWebSocket() {
    if (_isDisposed) return;

    final wsUrl = _baseUrl
        .replaceFirst('http://', 'ws://')
        .replaceFirst('https://', 'wss://');

    _wsChannel = WebSocketChannel.connect(
      Uri.parse('$wsUrl/jobs/$jobId/stream'),
    );

    _wsChannel!.stream.listen(
      (data) {
        if (_isDisposed) return;
        try {
          final json = jsonDecode(data as String) as Map<String, dynamic>;
          final update = JobStatusUpdate.fromJson(json);
          _statusController?.add(update);

          if (update.status.isTerminal) {
            _stopWebSocket();
          }
        } catch (e) {
          // Ignore parse errors
        }
      },
      onError: (error) {
        if (!_isDisposed) {
          _statusController?.addError(error);
        }
      },
      onDone: _stopWebSocket,
    );
  }

  void _stopWebSocket() {
    _wsChannel?.sink.close();
    _wsChannel = null;
  }

  /// Poll for job result (for environments without WebSocket support)
  Future<JobResult<T>> poll({
    Duration interval = const Duration(milliseconds: 500),
    Duration timeout = const Duration(minutes: 5),
  }) async {
    if (_cachedResult?.status.isTerminal == true) {
      return _cachedResult!;
    }

    final endTime = DateTime.now().add(timeout);
    final client = _createHttpClient();

    try {
      while (DateTime.now().isBefore(endTime)) {
        final response = await client.get(
          Uri.parse('$_baseUrl/jobs/$jobId'),
          headers: _headers,
        );

        if (response.statusCode != 200) {
          throw SynorException(
            'Failed to poll job status',
            statusCode: response.statusCode,
          );
        }

        final json = jsonDecode(response.body) as Map<String, dynamic>;
        final result = JobResult<T>.fromJson(json, _resultParser);

        if (result.status.isTerminal) {
          _cachedResult = result;
          return result;
        }

        await Future.delayed(interval);
      }

      throw SynorException('Job polling timed out after $timeout');
    } finally {
      client.close();
    }
  }

  /// Wait for job completion with automatic strategy selection
  Future<JobResult<T>> wait({
    Duration timeout = const Duration(minutes: 5),
    bool useWebSocket = true,
  }) async {
    if (_cachedResult?.status.isTerminal == true) {
      return _cachedResult!;
    }

    if (useWebSocket) {
      try {
        final completer = Completer<JobResult<T>>();
        late StreamSubscription<JobStatusUpdate> subscription;

        subscription = statusUpdates.listen(
          (update) async {
            if (update.status.isTerminal && !completer.isCompleted) {
              final result = await poll(
                interval: Duration.zero,
                timeout: const Duration(seconds: 10),
              );
              completer.complete(result);
              await subscription.cancel();
            }
          },
          onError: (error) {
            if (!completer.isCompleted) {
              completer.completeError(error);
            }
          },
        );

        return await completer.future.timeout(
          timeout,
          onTimeout: () {
            subscription.cancel();
            throw SynorException('Job wait timed out after $timeout');
          },
        );
      } catch (e) {
        // Fall back to polling if WebSocket fails
        return poll(timeout: timeout);
      }
    }

    return poll(timeout: timeout);
  }

  /// Cancel the job
  Future<bool> cancel() async {
    final client = _createHttpClient();
    try {
      final response = await client.post(
        Uri.parse('$_baseUrl/jobs/$jobId/cancel'),
        headers: _headers,
      );

      if (response.statusCode == 200) {
        _cachedResult = JobResult<T>(
          jobId: jobId,
          status: JobStatus.cancelled,
        );
        return true;
      }
      return false;
    } finally {
      client.close();
    }
  }

  /// Get current job status
  Future<JobStatus> getStatus() async {
    final client = _createHttpClient();
    try {
      final response = await client.get(
        Uri.parse('$_baseUrl/jobs/$jobId/status'),
        headers: _headers,
      );

      if (response.statusCode != 200) {
        throw SynorException(
          'Failed to get job status',
          statusCode: response.statusCode,
        );
      }

      final json = jsonDecode(response.body) as Map<String, dynamic>;
      return JobStatus.fromString(json['status'] as String);
    } finally {
      client.close();
    }
  }

  /// Dispose resources
  void dispose() {
    _isDisposed = true;
    _stopWebSocket();
    _statusController?.close();
    _statusController = null;
  }

  // Creates an HTTP client - in a real app, use http package
  dynamic _createHttpClient() {
    // This is a placeholder - actual implementation uses http package
    throw UnimplementedError('HTTP client should be injected');
  }
}

/// Batch job operations
class JobBatch<T> {
  final List<Job<T>> jobs;

  JobBatch(this.jobs);

  /// Wait for all jobs to complete
  Future<List<JobResult<T>>> waitAll({
    Duration timeout = const Duration(minutes: 10),
  }) async {
    return Future.wait(
      jobs.map((job) => job.wait(timeout: timeout)),
    );
  }

  /// Wait for first job to complete
  Future<JobResult<T>> waitAny({
    Duration timeout = const Duration(minutes: 5),
  }) async {
    return Future.any(
      jobs.map((job) => job.wait(timeout: timeout)),
    );
  }

  /// Cancel all jobs
  Future<void> cancelAll() async {
    await Future.wait(jobs.map((job) => job.cancel()));
  }

  /// Dispose all job resources
  void dispose() {
    for (final job in jobs) {
      job.dispose();
    }
  }
}