package geospatial;

import static org.junit.Assert.*;

import java.util.List;
import java.util.Random;

import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.query.SqlFieldsQuery;
import org.apache.ignite.cache.query.annotations.QuerySqlField;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.junit.Test;
import org.locationtech.jts.geom.Coordinate;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.geom.GeometryFactory;

public class SpatialIndexTest {

	private static final String CACHE_NAME = "points";
	private static final String QUERY = "select _key, _val from mappoint where coords && 'POLYGON((0 0, 0 100, 100 100, 100 0, 0 0))'";
	private static final String GEOM_IDX = "MAPPOINT_COORDS_IDX";
	
	private final GeometryFactory geomFactory = new GeometryFactory();

	/**
	 * MapPoint with indexed coordinates.
	 */
	public static class MapPoint {
		/** Coordinates. */
		@QuerySqlField(index = true)
		private Geometry coords;

		/**
		 * @param coords Coordinates.
		 */
		private MapPoint(Geometry coords) {
			this.coords = coords;
		}

		@Override
		public String toString() {
			return String.format("MapPoint [coords=%s]", coords);
		}
	}

	@Test
	public void testLoadAndQuery() {
		try (var server = server()) {
			try (var client = client()) {
				var indexedCache = client.createCache(new CacheConfiguration<Long, MapPoint>()
						.setName(CACHE_NAME)
						.setIndexedTypes(Long.class, MapPoint.class));
				load(indexedCache);
				assertEquals(1_000, indexedCache.size());
				var result = indexedCache.query(new SqlFieldsQuery("explain " + QUERY)).getAll();
				var row = result.get(0);
				assertNotEquals("should use index", -1, ((String)row.get(0)).indexOf(GEOM_IDX));
				result = indexedCache.query(new SqlFieldsQuery(QUERY)).getAll();
				assertEquals(9, result.size());
			}
		}
	}
	
	@Test
	public void testLoadAndQueryNewClientAccess() {
		try (var server = server()) {
			try (var client = client()) {
				var indexedCache = client.createCache(new CacheConfiguration<Long, MapPoint>()
						.setName(CACHE_NAME)
						.setIndexedTypes(Long.class, MapPoint.class));
				load(indexedCache);
				assertEquals(1_000, indexedCache.size());
				queryIndexed(indexedCache);
			}
			
			try (var client = client()) {
				// just connect a new client will case a logged exception 
				// Caused by: java.lang.NullPointerException
				//   at org.apache.ignite.internal.processors.query.h2.opt.GeoSpatialUtils.createIndex(GeoSpatialUtils.java:63)
				
				// accessing the cache finally results in a fail
				// Caused by: class org.apache.ignite.IgniteCheckedException: Type with name 'MapPoint' already indexed in cache 'points'.
				//   at org.apache.ignite.internal.processors.query.GridQueryProcessor.registerCache0(GridQueryProcessor.java:2158)
				var cache = client.<Long,MapPoint>cache(CACHE_NAME);
				assertEquals(1_000, cache.size());
				queryIndexed(cache);
			} 
		} 
	}
	
	private void queryIndexed(IgniteCache<Long, MapPoint> cache) {
		var result = cache.query(new SqlFieldsQuery("explain " + QUERY)).getAll();
		var row = result.get(0);
		assertNotEquals("should use index", -1, ((String)row.get(0)).indexOf(GEOM_IDX));
		result = cache.query(new SqlFieldsQuery(QUERY)).getAll();
		assertEquals(9, result.size());
	}
	

	private void load(IgniteCache<Long, MapPoint> cache) {
		var rnd = new Random(1234);
		for (var id = 0L; id < 1_000; id++) {
			var x = rnd.nextInt(1000);
			var y = rnd.nextInt(1000);
			var point = geomFactory.createPoint(new Coordinate(x, y));
			cache.put(id, new MapPoint(point));
		}
	}
	
	private IgniteConfiguration config() {
		var cfg = new IgniteConfiguration();
		cfg.setPeerClassLoadingEnabled(true);
		cfg.setMetricsLogFrequency(0);
		cfg.setDiscoverySpi(
				new TcpDiscoverySpi()
					.setIpFinder(new TcpDiscoveryVmIpFinder()
							.setAddresses(List.of("127.0.0.1"))));
		return cfg;
	}

	private Ignite server() {
		return Ignition.start(config().setIgniteInstanceName("server"));
	}

	private Ignite client() {
		return Ignition.start(config().setIgniteInstanceName("client").setClientMode(true));
	}
}