lib/src/datastore_impl.dart

// Copyright (c) 2014, the Dart project authors.  Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.

library gcloud.datastore_impl;

import 'dart:async';

import 'package:http/http.dart' as http;

import '../datastore.dart' as datastore;
import '../common.dart' show Page;
import 'package:googleapis_beta/datastore/v1beta2.dart' as api;

class TransactionImpl implements datastore.Transaction {
  final String data;
  TransactionImpl(this.data);
}

class DatastoreImpl implements datastore.Datastore {
  static const List<String> SCOPES = const <String>[
      api.DatastoreApi.DatastoreScope,
      api.DatastoreApi.UserinfoEmailScope,
  ];

  final api.DatastoreApi _api;
  final String _project;

  DatastoreImpl(http.Client client, this._project)
      : _api = new api.DatastoreApi(client);

  api.Key _convertDatastore2ApiKey(datastore.Key key, {bool enforceId: true}) {
    var apiKey = new api.Key();

    apiKey.partitionId = new api.PartitionId()
        ..datasetId = _project
        ..namespace = key.partition.namespace;

    apiKey.path = key.elements.map((datastore.KeyElement element) {
      var part = new api.KeyPathElement();
      part.kind = element.kind;
      if (element.id is int) {
        part.id = '${element.id}';
      } else if (element.id is String) {
        part.name = element.id;
      } else if (enforceId) {
        throw new datastore.ApplicationError(
            'Error while encoding entity key: Using `null` as the id is not '
            'allowed.');
      }
      return part;
    }).toList();

    return apiKey;
  }

  static datastore.Key _convertApi2DatastoreKey(api.Key key) {
    var elements = key.path.map((api.KeyPathElement element) {
      if (element.id != null) {
        return new datastore.KeyElement(element.kind, int.parse(element.id));
      } else if (element.name != null) {
        return new datastore.KeyElement(element.kind, element.name);
      } else {
        throw new datastore.DatastoreError(
            'Invalid server response: Expected allocated name/id.');
      }
    }).toList();

    var partition;
    if (key.partitionId != null) {
      partition = new datastore.Partition(key.partitionId.namespace);
      // TODO: assert projectId.
    }
    return new datastore.Key(elements, partition: partition);
  }

  bool _compareApiKey(api.Key a, api.Key b) {
    if (a.path.length != b.path.length) return false;

    // FIXME(Issue #2): Is this comparison working correctly?
    if (a.partitionId != null) {
      if (b.partitionId == null) return false;
      if (a.partitionId.datasetId != b.partitionId.datasetId) return false;
      if (a.partitionId.namespace != b.partitionId.namespace) return false;
    } else {
      if (b.partitionId != null) return false;
    }

    for (int i = 0; i < a.path.length; i++) {
      if (a.path[i].id != b.path[i].id ||
          a.path[i].name != b.path[i].name ||
          a.path[i].kind != b.path[i].kind) return false;
    }
    return true;
  }

  static _convertApi2DatastorePropertyValue(api.Value value) {
    if (value.booleanValue != null)
      return value.booleanValue;
    else if (value.integerValue != null)
      return int.parse(value.integerValue);
    else if (value.doubleValue != null)
      return value.doubleValue;
    else if (value.stringValue != null)
      return value.stringValue;
    else if (value.dateTimeValue != null)
      return value.dateTimeValue;
    else if (value.blobValue != null)
      return new datastore.BlobValue(value.blobValueAsBytes);
    else if (value.keyValue != null)
      return _convertApi2DatastoreKey(value.keyValue);
    else if (value.listValue != null)
      // FIXME(Issue #3): Consistently handle exceptions.
      throw new Exception('Cannot have lists inside lists.');
    else if (value.blobKeyValue != null)
      throw new UnsupportedError('Blob keys are not supported.');
    else if (value.entityValue != null)
      throw new UnsupportedError('Entity values are not supported.');
    return null;
  }

  api.Value _convertDatastore2ApiPropertyValue(
      value, bool indexed, {bool lists: true}) {
    var apiValue = new api.Value()
        ..indexed = indexed;
    if (value == null) {
      return apiValue;
    } else if (value is bool) {
      return apiValue
          ..booleanValue = value;
    } else if (value is int) {
      return apiValue
          ..integerValue = '$value';
    } else if (value is double) {
      return apiValue
          ..doubleValue = value;
    } else if (value is String) {
      return apiValue
          ..stringValue = value;
    } else if (value is DateTime) {
      return apiValue
          ..dateTimeValue = value;
    } else if (value is datastore.BlobValue) {
      return apiValue
          ..blobValueAsBytes = value.bytes;
    } else if (value is datastore.Key) {
      return apiValue
          ..keyValue = _convertDatastore2ApiKey(value, enforceId: false);
    } else if (value is List) {
      if (!lists) {
        // FIXME(Issue #3): Consistently handle exceptions.
        throw new Exception('List values are not allowed.');
      }

      convertItem(i)
          => _convertDatastore2ApiPropertyValue(i, indexed, lists: false);

      return new api.Value()
          ..listValue = value.map(convertItem).toList();
    } else {
      throw new UnsupportedError(
          'Types ${value.runtimeType} cannot be used for serializing.');
    }
  }

  static _convertApi2DatastoreProperty(api.Property property) {
    if (property.booleanValue != null)
      return property.booleanValue;
    else if (property.integerValue != null)
      return int.parse(property.integerValue);
    else if (property.doubleValue != null)
      return property.doubleValue;
    else if (property.stringValue != null)
      return property.stringValue;
    else if (property.dateTimeValue != null)
      return property.dateTimeValue;
    else if (property.blobValue != null)
      return new datastore.BlobValue(property.blobValueAsBytes);
    else if (property.keyValue != null)
      return _convertApi2DatastoreKey(property.keyValue);
    else if (property.listValue != null)
      return
          property.listValue.map(_convertApi2DatastorePropertyValue).toList();
    else if (property.blobKeyValue != null)
      throw new UnsupportedError('Blob keys are not supported.');
    else if (property.entityValue != null)
      throw new UnsupportedError('Entity values are not supported.');
    return null;
  }

  api.Property _convertDatastore2ApiProperty(
      value, bool indexed, {bool lists: true}) {
    var apiProperty = new api.Property()
        ..indexed = indexed;
    if (value == null) {
      return null;
    } else if (value is bool) {
      return apiProperty
          ..booleanValue = value;
    } else if (value is int) {
      return apiProperty
          ..integerValue = '$value';
    } else if (value is double) {
      return apiProperty
          ..doubleValue = value;
    } else if (value is String) {
      return apiProperty
          ..stringValue = value;
    } else if (value is DateTime) {
      return apiProperty
          ..dateTimeValue = value;
    } else if (value is datastore.BlobValue) {
      return apiProperty
          ..blobValueAsBytes = value.bytes;
    } else if (value is datastore.Key) {
      return apiProperty
          ..keyValue = _convertDatastore2ApiKey(value, enforceId: false);
    } else if (value is List) {
      if (!lists) {
        // FIXME(Issue #3): Consistently handle exceptions.
        throw new Exception('List values are not allowed.');
      }
      convertItem(i)
          => _convertDatastore2ApiPropertyValue(i, indexed, lists: false);
      return new api.Property()..listValue = value.map(convertItem).toList();
    } else {
      throw new UnsupportedError(
          'Types ${value.runtimeType} cannot be used for serializing.');
    }
  }

  static datastore.Entity _convertApi2DatastoreEntity(api.Entity entity) {
    var unindexedProperties = new Set();
    var properties = {};

    if (entity.properties != null) {
      entity.properties.forEach((String name, api.Property property) {
        properties[name] = _convertApi2DatastoreProperty(property);
        if (property.indexed == false) {
          // TODO(Issue #$4): Should we support mixed indexed/non-indexed list
          // values?
          if (property.listValue != null) {
            if (property.listValue.length > 0) {
              var firstIndexed = property.listValue.first.indexed;
              for (int i = 1; i < property.listValue.length; i++) {
                if (property.listValue[i].indexed != firstIndexed) {
                  throw new Exception('Some list entries are indexed and some '
                      'are not. This is currently not supported.');
                }
              }
              if (firstIndexed == false) {
                unindexedProperties.add(name);
              }
            }
          } else {
            unindexedProperties.add(name);
          }
        }
      });
    }
    return new datastore.Entity(_convertApi2DatastoreKey(entity.key),
                                properties,
                                unIndexedProperties: unindexedProperties);
  }

  api.Entity _convertDatastore2ApiEntity(datastore.Entity entity,
                                         {bool enforceId: false}) {
    var apiEntity = new api.Entity();

    apiEntity.key  = _convertDatastore2ApiKey(entity.key, enforceId: enforceId);
    apiEntity.properties = {};
    if (entity.properties != null) {
      for (var key in entity.properties.keys) {
        var value = entity.properties[key];
        bool indexed = false;
        if (entity.unIndexedProperties != null) {
          indexed = !entity.unIndexedProperties.contains(key);
        }
        var property = _convertDatastore2ApiPropertyValue(value, indexed);
        apiEntity.properties[key] = property;
      }
    }
    return apiEntity;
  }

  static Map<datastore.FilterRelation, String> relationMapping = const {
    datastore.FilterRelation.LessThan: 'LESS_THAN',
    datastore.FilterRelation.LessThanOrEqual: 'LESS_THAN_OR_EQUAL',
    datastore.FilterRelation.Equal: 'EQUAL',
    datastore.FilterRelation.GreatherThan: 'GREATER_THAN',
    datastore.FilterRelation.GreatherThanOrEqual: 'GREATER_THAN_OR_EQUAL',
    // TODO(Issue #5): IN operator not supported currently.
  };

  api.Filter _convertDatastore2ApiFilter(datastore.Filter filter) {
    var pf = new api.PropertyFilter();
    var operator = relationMapping[filter.relation];
    // FIXME(Issue #5): Is this OK?
    if (filter.relation == datastore.FilterRelation.In) {
      operator = 'EQUAL';
    }

    if (operator == null) {
      throw new ArgumentError('Unknown filter relation: ${filter.relation}.');
    }
    pf.operator = operator;
    pf.property = new api.PropertyReference()..name = filter.name;

    // FIXME(Issue #5): Is this OK?
    var value = filter.value;
    if (filter.relation == datastore.FilterRelation.In) {
      if (value is List && value.length == 1) {
        value = value.first;
      } else {
        throw new ArgumentError('List values not supported');
      }
    }

    pf.value = _convertDatastore2ApiPropertyValue(value, true, lists: false);
    return new api.Filter()..propertyFilter = pf;
  }

  api.Filter _convertDatastoreAncestorKey2ApiFilter(datastore.Key key) {
    var pf = new api.PropertyFilter();
    pf.operator = 'HAS_ANCESTOR';
    pf.property = new api.PropertyReference()..name = '__key__';
    pf.value = new api.Value()
        ..keyValue = _convertDatastore2ApiKey(key, enforceId: true);
    return new api.Filter()..propertyFilter = pf;
  }

  api.Filter _convertDatastore2ApiFilters(List<datastore.Filter> filters,
                                          datastore.Key ancestorKey) {
    if ((filters == null || filters.length == 0) && ancestorKey == null) {
      return null;
    }

    var compFilter = new api.CompositeFilter();
    if (filters != null) {
      compFilter.filters = filters.map(_convertDatastore2ApiFilter).toList();
    }
    if (ancestorKey != null) {
      var filter = _convertDatastoreAncestorKey2ApiFilter(ancestorKey);
      if (compFilter.filters == null) {
        compFilter.filters = [filter];
      } else {
        compFilter.filters.add(filter);
      }
    }
    compFilter.operator = 'AND';
    return new api.Filter()..compositeFilter = compFilter;
  }

  api.PropertyOrder _convertDatastore2ApiOrder(datastore.Order order) {
    var property = new api.PropertyReference()..name = order.propertyName;
    var direction = order.direction == datastore.OrderDirection.Ascending
        ? 'ASCENDING' : 'DESCENDING';
    return new api.PropertyOrder()
        ..direction = direction
        ..property = property;
  }

  List<api.PropertyOrder> _convertDatastore2ApiOrders(
      List<datastore.Order> orders) {
    if (orders == null) return null;

    return orders.map(_convertDatastore2ApiOrder).toList();
  }

  static Future _handleError(error, stack) {
    if (error is api.DetailedApiRequestError) {
      if (error.status == 400) {
        return new Future.error(
            new datastore.ApplicationError(error.message), stack);
      } else if (error.status == 409) {
        // NOTE: This is reported as:
        // "too much contention on these datastore entities"
        // TODO:
        return new Future.error(new datastore.TransactionAbortedError(), stack);
      } else if (error.status == 412) {
        return  new Future.error(new datastore.NeedIndexError(), stack);
      }
    }
    return new Future.error(error, stack);
  }

  Future<List<datastore.Key>> allocateIds(List<datastore.Key> keys) {
    var request = new api.AllocateIdsRequest();
    request..keys = keys.map((key) {
      return _convertDatastore2ApiKey(key, enforceId: false);
    }).toList();
    return _api.datasets.allocateIds(request, _project).then((response) {
      return response.keys.map(_convertApi2DatastoreKey).toList();
    }, onError: _handleError);
  }

  Future<datastore.Transaction> beginTransaction(
      {bool crossEntityGroup: false}) {
    var request = new api.BeginTransactionRequest();
    // TODO: Should this be made configurable?
    request.isolationLevel = 'SERIALIZABLE';
    return _api.datasets.beginTransaction(request, _project).then((result) {
      return new TransactionImpl(result.transaction);
    }, onError: _handleError);
  }

  Future<datastore.CommitResult> commit({List<datastore.Entity> inserts,
                                         List<datastore.Entity> autoIdInserts,
                                         List<datastore.Key> deletes,
                                         datastore.Transaction transaction}) {
    var request = new api.CommitRequest();

    if (transaction != null) {
      request.mode = 'TRANSACTIONAL';
      request.transaction = (transaction as TransactionImpl).data;
    } else {
      request.mode = 'NON_TRANSACTIONAL';
    }

    request.mutation = new api.Mutation();
    if (inserts != null) {
      request.mutation.upsert = new List(inserts.length);
      for (int i = 0; i < inserts.length; i++) {
        request.mutation.upsert[i] = _convertDatastore2ApiEntity(inserts[i]);
      }
    }
    if (autoIdInserts != null) {
      request.mutation.insertAutoId = new List(autoIdInserts.length);
      for (int i = 0; i < autoIdInserts.length; i++) {
        request.mutation.insertAutoId[i] =
            _convertDatastore2ApiEntity(autoIdInserts[i], enforceId: false);
      }
    }
    if (deletes != null) {
      request.mutation.delete = new List(deletes.length);
      for (int i = 0; i < deletes.length; i++) {
        request.mutation.delete[i] =
            _convertDatastore2ApiKey(deletes[i], enforceId: true);
      }
    }
    return _api.datasets.commit(request, _project).then((result) {
      var keys;
      if (autoIdInserts != null && autoIdInserts.length > 0) {
        keys = result
          .mutationResult
          .insertAutoIdKeys
          .map(_convertApi2DatastoreKey).toList();
      }
      return new datastore.CommitResult(keys);
    }, onError: _handleError);
  }

  Future<List<datastore.Entity>> lookup(List<datastore.Key> keys,
                                        {datastore.Transaction transaction}) {
    var apiKeys = keys.map((key) {
      return _convertDatastore2ApiKey(key, enforceId: true);
    }).toList();
    var request = new api.LookupRequest();
    request.keys = apiKeys;
    if (transaction != null) {
      // TODO: Make readOptions more configurable.
      request.readOptions = new api.ReadOptions();
      request.readOptions.transaction = (transaction as TransactionImpl).data;
    }
    return _api.datasets.lookup(request, _project).then((response) {
      if (response.deferred != null && response.deferred.length > 0) {
        throw new datastore.DatastoreError(
            'Could not successfully look up all keys due to resource '
            'constraints.');
      }

      // NOTE: This is worst-case O(n^2)!
      // Maybe we can optimize this somehow. But the API says:
      //  message LookupResponse {
      //    // The order of results in these fields is undefined and has no relation to
      //    // the order of the keys in the input.
      //
      //    // Entities found as ResultType.FULL entities.
      //    repeated EntityResult found = 1;
      //
      //    // Entities not found as ResultType.KEY_ONLY entities.
      //    repeated EntityResult missing = 2;
      //
      //    // A list of keys that were not looked up due to resource constraints.
      //    repeated Key deferred = 3;
      //  }
      var entities = new List(apiKeys.length);
      for (int i = 0; i < apiKeys.length; i++) {
        var apiKey = apiKeys[i];

        bool found = false;

        if (response.found != null) {
          for (var result in response.found) {
            if (_compareApiKey(apiKey, result.entity.key)) {
              entities[i] = _convertApi2DatastoreEntity(result.entity);
              found = true;
              break;
            }
          }
        }

        if (found) continue;

        if (response.missing != null) {
          for (var result in response.missing) {
            if (_compareApiKey(apiKey, result.entity.key)) {
              entities[i] = null;
              found = true;
              break;
            }
          }
        }

        if (!found) {
          throw new datastore.DatastoreError('Invalid server response: '
              'Tried to lookup ${apiKey.toJson()} but entity was neither in '
              'missing nor in found.');
        }
      }
      return entities;
    }, onError: _handleError);
  }

  Future<Page<datastore.Entity>> query(
      datastore.Query query, {datastore.Partition partition,
                              datastore.Transaction transaction}) {
    // NOTE: We explicitly do not set 'limit' here, since this is handled by
    // QueryPageImpl.runQuery.
    var apiQuery = new api.Query()
        ..filter = _convertDatastore2ApiFilters(query.filters,
                                                query.ancestorKey)
        ..order = _convertDatastore2ApiOrders(query.orders)
        ..offset = query.offset;

    if (query.kind != null) {
      apiQuery.kinds = [new api.KindExpression()..name = query.kind];
    }

    var request = new api.RunQueryRequest();
    request.query = apiQuery;
    if (transaction != null) {
      // TODO: Make readOptions more configurable.
      request.readOptions = new api.ReadOptions();
      request.readOptions.transaction = (transaction as TransactionImpl).data;
    }
    if (partition != null) {
      request.partitionId = new api.PartitionId()
          ..namespace = partition.namespace;
    }

    return QueryPageImpl.runQuery(_api, _project, request, query.limit)
        .catchError(_handleError);
  }

  Future rollback(datastore.Transaction transaction) {
    // TODO: Handle [transaction]
    var request = new api.RollbackRequest()
        ..transaction = (transaction as TransactionImpl).data;
    return _api.datasets.rollback(request, _project).catchError(_handleError);
  }
}

class QueryPageImpl implements Page<datastore.Entity> {
  static const int MAX_ENTITIES_PER_RESPONSE = 2000;

  final api.DatastoreApi _api;
  final String _project;
  final api.RunQueryRequest _nextRequest;
  final List<datastore.Entity> _entities;
  final bool _isLast;

  // This might be `null` in which case we request as many as we can get.
  final int _remainingNumberOfEntities;

  QueryPageImpl(this._api, this._project,
                this._nextRequest, this._entities,
                this._isLast, this._remainingNumberOfEntities);

  static Future<QueryPageImpl> runQuery(api.DatastoreApi api,
                                        String project,
                                        api.RunQueryRequest request,
                                        int limit,
                                        {int batchSize}) {
    int batchLimit = batchSize;
    if (batchLimit == null) {
      batchLimit = MAX_ENTITIES_PER_RESPONSE;
    }
    if (limit != null && limit < batchLimit) {
      batchLimit = limit;
    }

    request.query.limit = batchLimit;

    return api.datasets.runQuery(request, project).then((response) {
      var returnedEntities = const [];

      var batch = response.batch;
      if (batch.entityResults != null) {
        returnedEntities = batch.entityResults
            .map((result) => result.entity)
            .map(DatastoreImpl._convertApi2DatastoreEntity)
            .toList();
      }

      // This check is only necessary for the first request/response pair
      // (if offset was supplied).
      if (request.query.offset != null &&
          request.query.offset > 0 &&
          request.query.offset != response.batch.skippedResults) {
        throw new datastore.DatastoreError(
            'Server did not skip over the specified ${request.query.offset} '
            'entities.');
      }

      if (limit != null && returnedEntities.length > limit) {
        throw new datastore.DatastoreError(
            'Server returned more entities then the limit for the request'
            '(${request.query.limit}) was.');
      }


      // FIXME: TODO: Big hack!
      // It looks like Apiary/Atlas is currently broken.
      /*
      if (limit != null &&
          returnedEntities.length < batchLimit &&
          response.batch.moreResults == 'MORE_RESULTS_AFTER_LIMIT') {
        throw new datastore.DatastoreError(
            'Server returned response with less entities then the limit was, '
            'but signals there are more results after the limit.');
      }
      */

      // In case a limit was specified, we need to subtraction the number of
      // entities we already got.
      // (the checks above guarantee that this subraction is >= 0).
      int remainingEntities;
      if (limit != null) {
        remainingEntities = limit - returnedEntities.length;
      }

      // If the server signals there are more entities and we either have no
      // limit or our limit has not been reached, we set `moreBatches` to
      // `true`.
      bool moreBatches =
          (remainingEntities == null || remainingEntities > 0) &&
          response.batch.moreResults == 'MORE_RESULTS_AFTER_LIMIT';

      bool gotAll = limit != null && remainingEntities == 0;
      bool noMore = response.batch.moreResults == 'NO_MORE_RESULTS';
      bool isLast = gotAll || noMore;

      // As a sanity check, we assert that `moreBatches XOR isLast`.
      assert (isLast != moreBatches);

      // FIXME: TODO: Big hack!
      // It looks like Apiary/Atlas is currently broken.
      if (moreBatches && returnedEntities.length == 0) {
        print('Warning: Api to Google Cloud Datastore returned bogus response. '
              'Trying a workaround.');
        isLast = true;
        moreBatches = false;
      }

      if (!isLast && response.batch.endCursor == null) {
        throw new datastore.DatastoreError(
            'Server did not supply an end cursor, even though the query '
            'is not done.');
      }

      if (isLast) {
        return new QueryPageImpl(
            api, project, request, returnedEntities, true, null);
      } else {
        // NOTE: We reuse the old RunQueryRequest object here .

        // The offset will be 0 from now on, since the first request will have
        // skipped over the first `offset` results.
        request.query.offset = 0;

        // Furthermore we set the startCursor to the endCursor of the previous
        // result batch, so we can continue where we left off.
        request.query.startCursor = batch.endCursor;

        return new QueryPageImpl(
            api, project, request, returnedEntities, false, remainingEntities);
      }
    });
  }

  bool get isLast => _isLast;

  List<datastore.Entity> get items => _entities;

  Future<Page<datastore.Entity>> next({int pageSize}) {
    // NOTE: We do not respect [pageSize] here, the only mechanism we can
    // really use is `query.limit`, but this is user-specified when making
    // the query.
    if (isLast) {
      return new Future.sync(() {
        throw new ArgumentError('Cannot call next() on last page.');
      });
    }

    return QueryPageImpl.runQuery(
        _api, _project, _nextRequest, _remainingNumberOfEntities)
        .catchError(DatastoreImpl._handleError);
  }
}